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

import static java.util.Objects.requireNonNull;

import java.io.IOException;
import java.nio.charset.CharacterCodingException;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import javax.xml.datatype.XMLGregorianCalendar;

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.BatchWriter;
import org.apache.accumulo.core.client.Connector;
import org.apache.accumulo.core.client.MultiTableBatchWriter;
import org.apache.accumulo.core.client.MutationsRejectedException;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.ScannerBase;
import org.apache.accumulo.core.client.TableExistsException;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.commons.codec.binary.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Logger;
import org.apache.rya.accumulo.experimental.AbstractAccumuloIndexer;
import org.apache.rya.api.RdfCloudTripleStoreConfiguration;
import org.apache.rya.api.client.RyaClientException;
import org.apache.rya.api.domain.RyaStatement;
import org.apache.rya.api.resolver.RyaToRdfConversions;
import org.apache.rya.indexing.KeyParts;
import org.apache.rya.indexing.StatementConstraints;
import org.apache.rya.indexing.StatementSerializer;
import org.apache.rya.indexing.TemporalIndexer;
import org.apache.rya.indexing.TemporalInstant;
import org.apache.rya.indexing.TemporalInstantRfc3339;
import org.apache.rya.indexing.TemporalInterval;
import org.apache.rya.indexing.accumulo.ConfigUtils;
import org.eclipse.rdf4j.common.iteration.CloseableIteration;
import org.eclipse.rdf4j.model.IRI;
import org.eclipse.rdf4j.model.Literal;
import org.eclipse.rdf4j.model.Resource;
import org.eclipse.rdf4j.model.Statement;
import org.eclipse.rdf4j.query.QueryEvaluationException;
import org.joda.time.DateTime;

public class AccumuloTemporalIndexer extends AbstractAccumuloIndexer implements TemporalIndexer {

    private static final String TABLE_SUFFIX = "temporal";

    private static final Logger logger = Logger.getLogger(AccumuloTemporalIndexer.class);

    private static final String CF_INTERVAL = "interval";


    // Delimiter used in the interval stored in the triple's object literal.
    // So far, no ontology specifies a date range, just instants.
    // Set to the same delimiter used by the indexer, probably needs revisiting.
    //private static final String REGEX_intervalDelimiter = TemporalInterval.DELIMITER;

    private Configuration conf;

    private MultiTableBatchWriter mtbw;

    private BatchWriter temporalIndexBatchWriter;

    private Set<IRI> validPredicates;
    private String temporalIndexTableName;

    private boolean isInit = false;


    /**
     * intilize the temporal index.
     * This is dependent on a few set method calls before init:
     * >  Connector = ConfigUtils.getConnector(conf);
     * >  MultiTableBatchWriter mtbw = connector.createMultiTableBatchWriter(new BatchWriterConfig());
     * >  // optional: temporal.setConnector(connector);
     * >  temporal.setMultiTableBatchWriter(mtbw);
     * >  temporal.init();
     */
    @Override
    public void init() {
        if (!isInit) {
            try {
                initReadWrite();
                isInit = true;
            } catch (final AccumuloException | AccumuloSecurityException | TableNotFoundException | TableExistsException | RyaClientException e) {
                logger.error("Unable to initialize index.  Throwing Runtime Exception. ", e);
                throw new RuntimeException(e);
            }
        }
    }
    /**
     * Initialize for writable use.  
     * This is called from the DAO, perhaps others.
     */
    private void initReadWrite() throws AccumuloException, AccumuloSecurityException, TableNotFoundException,
                    TableExistsException, RyaClientException {
        if (mtbw == null)
            throw new RyaClientException("Failed to initialize temporal index, setMultiTableBatchWriter() was not set.");
        if (conf == null)
            throw new RyaClientException("Failed to initialize temporal index, setConf() was not set.");
        if (temporalIndexTableName==null)
            throw new RyaClientException("Failed to set temporalIndexTableName==null.");

        // Now do all the writable setup, read should already be complete.
        // Create one index table on first run.
        Boolean isCreated = ConfigUtils.createTableIfNotExists(conf, temporalIndexTableName);
        if (isCreated) {
            logger.info("First run, created temporal index table: " + temporalIndexTableName);
        }
        temporalIndexBatchWriter = mtbw.getBatchWriter(temporalIndexTableName);
    }

    /**
     * Initialize everything for a query-only use.  
     * This is called from setConf, since that must be called by anyone.
     * The DAO will also call setMultiTableBatchWriter() and init().
     */
	private void initReadOnly()  {
		if (conf == null)
			throw new Error("Failed to initialize temporal index, setConf() was not set.");
		temporalIndexTableName = getTableName();
		validPredicates = ConfigUtils.getTemporalPredicates(conf);
	}

    /**
     * Set the configuration, then initialize for read (query) use only.
     * Readonly initialization occurs in setConf because it does not require setting a multitablebatchwriter (mtbw).
     */
    @Override
    public void setConf(final Configuration conf) {
        this.conf = conf;
			initReadOnly();
    }

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


    /**
     * Store a statement in the index if it meets the criterion: Object should be
     * a literal and one of the validPredicates from the configuration.
     * If it does not meet the criteria, it is silently ignored.
     * logs a warning if the object is not parse-able.
     * Attempts to parse with calendarValue = literalValue.calendarValue()
     * if that fails, tries: org.joda.time.DateTime.parse() .
     * T O D O parse an interval using multiple predicates for same subject -- ontology dependent.
     */
    private void storeStatement(final Statement statement) throws IOException, IllegalArgumentException {
    	Objects.requireNonNull(temporalIndexBatchWriter,"This is not initialized for writing.  Must call setMultiTableBatchWriter() and init().");
        // if the predicate list is empty, accept all predicates.
        // Otherwise, make sure the predicate is on the "valid" list
        final boolean isValidPredicate = validPredicates == null || validPredicates.isEmpty() || validPredicates.contains(statement.getPredicate());
        if (!isValidPredicate || !(statement.getObject() instanceof Literal)) {
            return;
        }

        final DateTime[] indexDateTimes = new DateTime[2]; // 0 begin, 1 end of interval
        extractDateTime(statement, indexDateTimes);
        if (indexDateTimes[0]==null) {
            return;
        }

        if (!this.isInit)
            throw new RuntimeException("Method .init() was not called (or failed) before attempting to store statements.");

        // Add this as an instant, or interval.
        try {
            if (indexDateTimes[1] != null) {
                final TemporalInterval interval = new TemporalInterval(new TemporalInstantRfc3339(indexDateTimes[0]), new TemporalInstantRfc3339(indexDateTimes[1]));
                addInterval(temporalIndexBatchWriter, interval, statement);
            } else {
                final TemporalInstant instant = new TemporalInstantRfc3339(indexDateTimes[0]);
                addInstant(temporalIndexBatchWriter, instant, statement);
            }
        } catch (final MutationsRejectedException e) {
            throw new IOException("While adding interval/instant for statement =" + statement, e);
        }
    }


    @Override
    public void storeStatement(final RyaStatement statement) throws IllegalArgumentException, IOException {
        storeStatement(RyaToRdfConversions.convertStatement(statement));
    }


    /**
     * parse the literal dates from the object of a statement.
     *
     * @param statement
     * @param outputDateTimes
     */
    private void extractDateTime(final Statement statement, final DateTime[] outputDateTimes) {
        if (!(statement.getObject() instanceof Literal)) {
            throw new RuntimeException("Statement's object must be a literal: " + statement);
        }
        // throws IllegalArgumentException NumberFormatException if can't parse
        String logThis = null;         final Literal literalValue = (Literal) statement.getObject();
        // First attempt to parse a interval in the form "[date1,date2]"
        final Matcher matcher = Pattern.compile("\\[(.*)\\,(.*)\\].*").matcher(literalValue.stringValue());
        if (matcher.find()) {
            try {
                // Got a datetime pair, parse into an interval.
                outputDateTimes[0] = new DateTime(matcher.group(1));
                outputDateTimes[1] = new DateTime(matcher.group(2));
                return;
            } catch (final java.lang.IllegalArgumentException e) {
                logThis = e.getMessage() + " " + logThis;
                outputDateTimes[0]=null;
                outputDateTimes[1]=null;
            }
        }

        try {
            final XMLGregorianCalendar calendarValue = literalValue.calendarValue();
            outputDateTimes[0] = new DateTime(calendarValue.toGregorianCalendar());
            outputDateTimes[1] = null;
            return;
        } catch (final java.lang.IllegalArgumentException e) {
            logThis = e.getMessage();
        }
        // Try again using Joda Time DateTime.parse()
        try {
            outputDateTimes[0] = DateTime.parse(literalValue.stringValue());
            outputDateTimes[1] = null;
            //System.out.println(">>>>>>>Joda parsed: "+literalValue.stringValue());
            return;
        } catch (final java.lang.IllegalArgumentException e) {
            logThis = e.getMessage() + " " + logThis;
        }
        logger.warn("TemporalIndexer is unable to parse the date/time from statement="  + statement.toString() + " " +logThis);
        return;
    }

    /**
     * Remove an interval index
     * TODO: integrate into KeyParts (or eliminate)
     * @param writer
     * @param cv
     * @param interval
     * @throws MutationsRejectedException
     */
    public void removeInterval(final BatchWriter writer, final TemporalInterval interval, final Statement statement) throws MutationsRejectedException {
        final Text cf = new Text(StatementSerializer.writeContext(statement));
        final Text cqBegin = new Text(KeyParts.CQ_BEGIN);
        final Text cqEnd = new Text(KeyParts.CQ_END);

        // Start Begin index
        Text keyText = new Text(interval.getAsKeyBeginning());
        KeyParts.appendUniqueness(statement, keyText);
        Mutation m = new Mutation(keyText);
        m.putDelete(cf, cqBegin);
        writer.addMutation(m);

        // now the end index:
        keyText = new Text(interval.getAsKeyEnd());
        KeyParts.appendUniqueness(statement, keyText);
        m = new Mutation(keyText);
        m.putDelete(cf, cqEnd);
        writer.addMutation(m);
    }

    /**
     * Remove an interval instant
     *
     * @param writer
     * @param cv
     * @param instant
     * @throws MutationsRejectedException
     */
    public void removeInstant(final BatchWriter writer, final TemporalInstant instant, final Statement statement) throws MutationsRejectedException {
        final KeyParts keyParts = new KeyParts(statement, instant);
        for (final KeyParts  k: keyParts) {
            final Mutation m = new Mutation(k.getStoreKey());
            m.putDelete(k.cf, k.cq);
            writer.addMutation(m);
        }
    }

    /**
     * Index a new interval
     * TODO: integrate into KeyParts (or eliminate)
     * @param writer
     * @param cv
     * @param interval
     * @throws MutationsRejectedException
     */
    public void addInterval(final BatchWriter writer, final TemporalInterval interval, final Statement statement) throws MutationsRejectedException {

        final Value statementValue = new Value(StringUtils.getBytesUtf8(StatementSerializer.writeStatement(statement)));
        final Text cf = new Text(StatementSerializer.writeContext(statement));
        final Text cqBegin = new Text(KeyParts.CQ_BEGIN);
        final Text cqEnd = new Text(KeyParts.CQ_END);

        // Start Begin index
        Text keyText = new Text(interval.getAsKeyBeginning());
        KeyParts.appendUniqueness(statement, keyText);
        Mutation m = new Mutation(keyText);
        m.put(cf, cqBegin, statementValue);
        // System.out.println("mutations add begin row=" + m.getRow() + " value=" + value.toString());
        writer.addMutation(m);

        // now the end index:
        keyText = new Text(interval.getAsKeyEnd());
        KeyParts.appendUniqueness(statement, keyText);
        m = new Mutation(keyText);
        m.put(cf, cqEnd, new Value(statementValue));
        // System.out.println("mutations add end row=" + m.getRow() + " value=" + value.toString());
        writer.addMutation(m);
    }


    /**
     * Index a new instant
     * Make indexes that handle this expression:
     *     hash( s? p? ) ?o
     *         == o union hash(s)o union hash(p)o  union hash(sp)o
     *
     * @param writer
     * @param cv
     * @param instant
     * @throws MutationsRejectedException
     */
    public void addInstant(final BatchWriter writer, final TemporalInstant instant, final Statement statement) throws MutationsRejectedException {
        final KeyParts keyParts = new KeyParts(statement, instant);
        for (final KeyParts k : keyParts) {
            final Mutation m = new Mutation(k.getStoreKey());
            m.put(k.cf, k.cq,k.getValue());
            writer.addMutation(m);
        }
    }


    /**
     * creates a scanner and handles all the throwables and nulls.
     *
     * @param scanner
     * @return
     * @throws IOException
     */
    private Scanner getScanner() throws QueryEvaluationException {
        final String whileDoing = "While creating a scanner for a temporal query. table name=" + temporalIndexTableName;
        Scanner scanner = null;
        try {
            scanner = ConfigUtils.createScanner(temporalIndexTableName, conf);
        } catch (final AccumuloException e) {
            logger.error(whileDoing, e);
            throw new QueryEvaluationException(whileDoing, e);
        } catch (final AccumuloSecurityException e) {
            throw new QueryEvaluationException(whileDoing, e);
        } catch (final TableNotFoundException e) {
            logger.error(whileDoing, e);
            throw new QueryEvaluationException(whileDoing
                    + " The temporal index table should have been created by this constructor, if found missing.", e);
        }
        return scanner;
    }

    private BatchScanner getBatchScanner() throws QueryEvaluationException {
        final String whileDoing = "While creating a Batch scanner for a temporal query. table name=" + temporalIndexTableName;
        try {
            return ConfigUtils.createBatchScanner(temporalIndexTableName, conf);
        } catch (final AccumuloException e) {
            logger.error(whileDoing, e);
            throw new QueryEvaluationException(whileDoing, e);
        } catch (final AccumuloSecurityException e) {
            throw new QueryEvaluationException(whileDoing, e);
        } catch (final TableNotFoundException e) {
            logger.error(whileDoing, e);
            throw new QueryEvaluationException(whileDoing
                    + " The temporal index table should have been created by this constructor, if found missing. ", e);
        }
    }

    /**
     * statements where the datetime is exactly the same as the queryInstant.
     */
    @Override
    public CloseableIteration<Statement, QueryEvaluationException> queryInstantEqualsInstant(
            final TemporalInstant queryInstant, final StatementConstraints constraints)
            throws QueryEvaluationException {
            // get rows where the repository time is equal to the given time in queryInstant.
            final Query query = new Query() {
                @Override
                public Range getRange(final KeyParts keyParts) {
                    //System.out.println("Scanning queryInstantEqualsInstant: prefix:" + KeyParts.toHumanString(keyParts.getQueryKey()));
                    return Range.prefix(keyParts.getQueryKey()); // <-- specific logic
                }
            };
            final ScannerBase scanner = query.doQuery(queryInstant, constraints);
            // TODO currently context constraints are filtered on the client.
            return getContextIteratorWrapper(scanner, constraints.getContext());
    }

    /**
     * get statements where the db row ID is BEFORE the given queryInstant.
     */
    @Override
    public CloseableIteration<Statement, QueryEvaluationException> queryInstantBeforeInstant(
            final TemporalInstant queryInstant, final StatementConstraints constraints)
            throws QueryEvaluationException {
        // get rows where the repository time is before the given time.
        final Query query = new Query() {
            @Override
            public Range getRange(final KeyParts keyParts) {
                Text start= null;
                if (keyParts.constraintPrefix != null ) {
                    start = keyParts.constraintPrefix;   // <-- start specific logic
                } else {
                    start = new Text(KeyParts.HASH_PREFIX_FOLLOWING);
                }
                final Text endAt = keyParts.getQueryKey();                   // <-- end specific logic
                //System.out.println("Scanning queryInstantBeforeInstant: from:" + KeyParts.toHumanString(start) + " up to:" + KeyParts.toHumanString(endAt));
                return new Range(start, true, endAt, false);
            }
        };
        final ScannerBase scanner = query.doQuery(queryInstant, constraints);
        return getContextIteratorWrapper(scanner, constraints.getContext());
    }

    /**
     * get statements where the date object is after the given queryInstant.
     */
    @Override
    public CloseableIteration<Statement, QueryEvaluationException> queryInstantAfterInstant(
            final TemporalInstant queryInstant, final StatementConstraints constraints)
            throws QueryEvaluationException {
        final Query query = new Query() {
            @Override
            public Range getRange(final KeyParts keyParts) {
                final Text start = Range.followingPrefix(keyParts.getQueryKey());  // <-- specific logic
                Text endAt = null;  // no constraints                            // <-- specific logic
                if (keyParts.constraintPrefix != null ) {
                    endAt = Range.followingPrefix(keyParts.constraintPrefix);
                }
                //System.out.println("Scanning queryInstantAfterInstant from after:" + KeyParts.toHumanString(start) + " up to:" + KeyParts.toHumanString(endAt));
                return new Range(start, true, endAt, false);
            }
        };
        final ScannerBase scanner = query.doQuery(queryInstant, constraints);
        return getContextIteratorWrapper(scanner, constraints.getContext());
    }

    /**
     * Get instances before a given interval. Returns queryInstantBeforeInstant with the interval's beginning time.
     */
    @Override
    public CloseableIteration<Statement, QueryEvaluationException> queryInstantBeforeInterval(
            final TemporalInterval givenInterval, final StatementConstraints contraints)
            throws QueryEvaluationException {
        return queryInstantBeforeInstant(givenInterval.getHasBeginning(), contraints);
    }

    /**
     * Get instances after a given interval. Returns queryInstantAfterInstant with the interval's end time.
     */
    @Override
    public CloseableIteration<Statement, QueryEvaluationException> queryInstantAfterInterval(
            final TemporalInterval givenInterval, final StatementConstraints contraints) throws QueryEvaluationException {
        return queryInstantAfterInstant(givenInterval.getHasEnd(), contraints);
    }

    /**
     * Get instances inside a given interval.
     * Returns after interval's beginning time, and before ending time,
     * exclusive (don't match the beginning and ending).
     */
    @Override
    public CloseableIteration<Statement, QueryEvaluationException> queryInstantInsideInterval(
            final TemporalInterval queryInterval, final StatementConstraints constraints)
            throws QueryEvaluationException {
        // get rows where the time is after the given interval's beginning time and before the ending time.
        final TemporalInterval theQueryInterval = queryInterval;
        final Query query = new Query() {
            private final TemporalInterval queryInterval = theQueryInterval;
            @Override
            public Range getRange(final KeyParts keyParts) {
                final Text start = Range.followingPrefix(new Text(keyParts.getQueryKey(queryInterval.getHasBeginning())));
                final Text endAt = new Text(keyParts.getQueryKey(queryInterval.getHasEnd())); // <-- end specific logic
                //System.out.println("Scanning queryInstantInsideInterval: from excluding:" + KeyParts.toHumanString(start) + " up to:" + KeyParts.toHumanString(endAt));
                return new Range(start, false, endAt, false);
            }
        };
        final ScannerBase scanner = query.doQuery(queryInterval.getHasBeginning(), constraints);
        return getContextIteratorWrapper(scanner, constraints.getContext());
    }
    /**
     * Get instances matching the beginning of a given interval.
     */
    @Override
    public CloseableIteration<Statement, QueryEvaluationException> queryInstantHasBeginningInterval(
            final TemporalInterval queryInterval, final StatementConstraints contraints)
            throws QueryEvaluationException {
        return queryInstantEqualsInstant(queryInterval.getHasBeginning(), contraints);
    }

    /**
     * Get instances matching the ending of a given interval.
     */
    @Override
    public CloseableIteration<Statement, QueryEvaluationException> queryInstantHasEndInterval(
            final TemporalInterval queryInterval, final StatementConstraints contraints)
            throws QueryEvaluationException {
        return queryInstantEqualsInstant(queryInterval.getHasEnd(), contraints);
    }

    /**
     * Get intervals stored in the repository matching the given interval.
     * Indexing Intervals  will probably change or be removed.
     * Currently predicate and subject constraints are filtered on the client.
     */
    @Override
    public CloseableIteration<Statement, QueryEvaluationException> queryIntervalEquals(
            final TemporalInterval query, final StatementConstraints contraints)
            throws QueryEvaluationException {
        final Scanner scanner = getScanner();
        if (scanner != null) {
            // get rows where the start and end match.
            final Range range = Range.prefix(new Text(query.getAsKeyBeginning()));
            scanner.setRange(range);
            if (contraints.hasContext()) {
                scanner.fetchColumn(new Text(contraints.getContext().toString()), new Text(KeyParts.CQ_BEGIN));
            } else {
                scanner.fetchColumn(new Text(""), new Text(KeyParts.CQ_BEGIN));
            }
        }
        // Iterator<Entry<Key, Value>> iter = scanner.iterator();
        // while (iter.hasNext()) {
        // System.out.println("queryIntervalEquals results:"+iter.next());
        // }
        //return getConstrainedIteratorWrapper(scanner, contraints);
        return getIteratorWrapper(scanner);
    }

    /**
     * find intervals stored in the repository before the given Interval. Find interval endings that are
     * before the given beginning.
     * Indexing Intervals  will probably change or be removed.
     * Currently predicate and subject constraints are filtered on the client.
     */
    @Override
    public CloseableIteration<Statement, QueryEvaluationException> queryIntervalBefore(
            final TemporalInterval queryInterval, final StatementConstraints constraints) throws QueryEvaluationException
    {
        final Scanner scanner = getScanner();
        if (scanner != null) {
            // get rows where the end date is less than the queryInterval.getBefore()
            final Range range = new Range(null, false, new Key(new Text(queryInterval.getHasBeginning().getAsKeyBytes())), false);
            scanner.setRange(range);
             if (constraints.hasContext()) {
                scanner.fetchColumn(new Text(constraints.getContext().toString()), new Text(KeyParts.CQ_END));
            } else {
                scanner.fetchColumn(new Text(""), new Text(KeyParts.CQ_END));
            }
        }
        return getIteratorWrapper(scanner);
    }

    /**
     * Interval after given interval.  Find intervals that begin after the endings of the given interval.
     * Use the special following prefix mechanism to avoid matching the beginning date.
     * Indexing Intervals  will probably change or be removed.
     * Currently predicate and subject and context constraints are filtered on the client.
     */
    @Override
    public CloseableIteration<Statement, QueryEvaluationException> queryIntervalAfter(
            final TemporalInterval queryInterval, final StatementConstraints constraints)
            throws QueryEvaluationException {

        final Scanner scanner = getScanner();
        if (scanner != null) {
            // get rows where the start date is greater than the queryInterval.getEnd()
            final Range range = new Range(new Key(Range.followingPrefix(new Text(queryInterval.getHasEnd().getAsKeyBytes()))), false, null, true);
            scanner.setRange(range);

            if (constraints.hasContext()) {
                scanner.fetchColumn(new Text(constraints.getContext().toString()), new Text(KeyParts.CQ_BEGIN));
            } else {
                scanner.fetchColumn(new Text(""), new Text(KeyParts.CQ_BEGIN));
            }
        }
        // TODO currently predicate, subject and context constraints are filtered on the clients
        return getIteratorWrapper(scanner);
    }
    // --
    // -- END of Query functions.  Next up, general stuff used by the queries above.
    // --

    /**
     * Allows passing range specific logic into doQuery.
     * Each query function implements an anonymous instance of this and calls it's doQuery().
     */
    abstract class Query {
        abstract protected Range getRange(KeyParts keyParts);

        public ScannerBase doQuery(final TemporalInstant queryInstant, final StatementConstraints constraints) throws QueryEvaluationException {
            // key is contraintPrefix + time, or just time.
            // Any constraints handled here, if the constraints are empty, the
            // thisKeyParts.contraintPrefix will be null.
            final List<KeyParts> keyParts = KeyParts.keyPartsForQuery(queryInstant, constraints);
            ScannerBase scanner = null;
            if (keyParts.size() > 1) {
                scanner = getBatchScanner();
            } else {
                scanner = getScanner();
            }

            final Collection<Range> ranges = new HashSet<Range>();
            KeyParts lastKeyParts = null;
            Range range = null;
            for (final KeyParts thisKeyParts : keyParts) {
                range = getRange(thisKeyParts);
                ranges.add(range);
                lastKeyParts = thisKeyParts;
            }
            if (lastKeyParts == null || scanner == null) {
                throw new NullPointerException("lastkeyParts or scanner is null, impossible! keyParts.size()= " + keyParts.size() + " scanner= " + scanner);
            }
            //System.out.println("Scanning columns, cf:" + lastKeyParts.cf + "CQ:" + lastKeyParts.cq);
            scanner.fetchColumn(new Text(lastKeyParts.cf), new Text(lastKeyParts.cq));
            if (scanner instanceof BatchScanner) {
                ((BatchScanner) scanner).setRanges(ranges);
            } else if (range != null) {
                ((Scanner) scanner).setRange(range);
            }
            return scanner;
        }
    }

    /**
     * An iteration wrapper for a loaded scanner that is returned for each query above.
     *
     * @param scanner
     *            the results to iterate, then close.
     * @return an anonymous object that will iterate the resulting statements from a given scanner.
     */
    private static CloseableIteration<Statement, QueryEvaluationException> getIteratorWrapper(final ScannerBase scanner) {

        final Iterator<Entry<Key, Value>> i = scanner.iterator();

        return new CloseableIteration<Statement, QueryEvaluationException>() {
            @Override
            public boolean hasNext() {
                return i.hasNext();
            }

            @Override
            public Statement next() throws QueryEvaluationException {
                final Entry<Key, Value> entry = i.next();
                final Value v = entry.getValue();
                try {
                    final String dataString = Text.decode(v.get(), 0, v.getSize());
                    final Statement s = StatementSerializer.readStatement(dataString);
                    return s;
                } catch (final CharacterCodingException e) {
                    logger.error("Error decoding value=" + Arrays.toString(v.get()), e);
                    throw new QueryEvaluationException(e);
                } catch (final IOException e) {
                    logger.error("Error de-serializing statement, string=" + v.get(), e);
                    throw new QueryEvaluationException(e);
                }
            }

            @Override
            public void remove() {
                throw new UnsupportedOperationException("Remove not implemented");
            }

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


    /**
     * An iteration wrapper for a loaded scanner that is returned for partially supported interval queries above.
     *
     * @param scanner  the results to iterate, then close.
     * @param constraints  limit statements returned by next() to those matching the constraints.
     * @return an anonymous object that will iterate the resulting statements from a given scanner.
     * @throws QueryEvaluationException
     */
    private static CloseableIteration<Statement, QueryEvaluationException> getConstrainedIteratorWrapper(final Scanner scanner, final StatementConstraints constraints) {
        if (!constraints.hasContext() && !constraints.hasSubject() && !constraints.hasPredicates()) {
            return getIteratorWrapper(scanner);
        }
        return new ConstrainedIteratorWrapper(scanner) {
            @Override
            public boolean allowedBy(final Statement statement) {
                return allowedByConstraints(statement, constraints);
            }
        };
    }
    /**
     * An iteration wrapper for a loaded scanner that is returned for queries above.
     * Currently, this temporal index supports contexts only on the client, using this filter.
     *
     * @param scanner  the results to iterate, then close.
     * @param constraints  limit statements returned by next() to those matching the constraints.
     * @return an anonymous object that will iterate the resulting statements from a given scanner.
     * @throws QueryEvaluationException
     */
    private static CloseableIteration<Statement, QueryEvaluationException> getContextIteratorWrapper(final ScannerBase scanner, final Resource context) {
        if (context==null) {
            return getIteratorWrapper(scanner);
        }
        return new ConstrainedIteratorWrapper(scanner) {
            @Override
            public boolean allowedBy(final Statement statement) {
                return allowedByContext(statement,  context);
            }
        };
    }
    /**
     * Wrap a scanner in a iterator that will filter statements based on a boolean allowedBy().
     * If the allowedBy function returns false for the next statement, it is skipped.
     * This is used for to do client side, what the index cannot (yet) do on the server side.
     */
    abstract static class ConstrainedIteratorWrapper implements CloseableIteration<Statement, QueryEvaluationException> {
            private Statement nextStatement=null;
            private boolean isInitialized = false;
            final private Iterator<Entry<Key, Value>> i;
            final private ScannerBase scanner;

            ConstrainedIteratorWrapper(final ScannerBase scanner) {
                this.scanner = scanner;
                i=scanner.iterator();
            }
            @Override
            public boolean hasNext() throws QueryEvaluationException {
                if (!isInitialized) {
                    internalGetNext();
                }
                return (nextStatement != null) ;
            }

            @Override
            public Statement next() throws QueryEvaluationException {
                if (nextStatement==null) {
                    if (!isInitialized) {
                        internalGetNext();
                    }
                    if (nextStatement==null) {
                        throw new NoSuchElementException();
                    }
                }
                // use this one, then get the next one loaded.
                final Statement thisStatement = nextStatement;
                internalGetNext();
                return thisStatement;
            }

            /**
             * Gets the next statement meeting constraints and stores in nextStatement.
             * Sets null when all done, or on exception.
             * @throws QueryEvaluationException
             */
            private void internalGetNext()
                    throws QueryEvaluationException {
                isInitialized=true;
                nextStatement = null;  // Default on done or error.
                Statement statement = null;
                while (i.hasNext()) {
                    final Entry<Key, Value> entry = i.next();
                    final Value v = entry.getValue();
                    try {
                        final String dataString = Text.decode(v.get(), 0, v.getSize());
                        statement = StatementSerializer.readStatement(dataString);
                    } catch (final CharacterCodingException e) {
                        logger.error("Error decoding value=" + Arrays.toString(v.get()), e);
                        throw new QueryEvaluationException(e);
                    } catch (final IOException e) {
                        logger.error("Error de-serializing statement, string=" + v.get(), e);
                        throw new QueryEvaluationException(e);
                    }
                    if (allowedBy(statement)) {
                        nextStatement = statement;
                        return;
                    }
                }
            }
            public abstract boolean allowedBy(Statement s);

            @Override
            public void remove() {
                throw new UnsupportedOperationException("Remove not implemented");
            }

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

    /**
     * Does the statement meet the constraints? Match predicate, subject, and context.
     * @param statement  Candidate statement to be allowed or not.
     * @param contraints  fields that are non-null must match the statement's components, otherwise it is not allowed.
     * @return true if the parts of the statement match the statementConstraints' parts.
     */
    protected static boolean allowedByConstraints(final Statement statement, final StatementConstraints constraints) {

        if (constraints.hasSubject() && ! constraints.getSubject().toString().equals(statement.getSubject().toString()))
            {System.out.println("Constrain subject: "+constraints.getSubject()+" != " + statement.getSubject()); return false;}
            //return false;

        if (! allowedByContext(statement, constraints.getContext()))
         {
            return false;
            //{System.out.println("Constrain context: "+constraints.getContext()+" != " + statement.getContext()); return false;}
        }

        if (constraints.hasPredicates() && ! constraints.getPredicates().contains(statement.getPredicate()))
         {
            return false;
            //{System.out.println("Constrain predicate: "+constraints.getPredicates()+" != " + statement.getPredicate()); return false;}
        }

        System.out.println("allow statement: "+ statement.toString());
        return true;
    }

    /**
     * Allow only if the context matches the statement.  This is a client side filter.
     * @param statement
     * @param context
     * @return
     */
    protected static boolean allowedByContext(final Statement statement, final Resource context) {
        return context==null || context.equals( statement.getContext() );
    }

    @Override
    public Set<IRI> getIndexablePredicates() {

        return validPredicates;
    }

    /**
     * Flush the data to the batchwriter.
     * Throws a IOException as required by the flushable interface,
     * wrapping MutationsRejectedException.
     */
    @Override
    public void flush() throws IOException {
        try {
            mtbw.flush();
        } catch (final MutationsRejectedException e) {
            final String msg = "Error while flushing the batch writer.";
            logger.error(msg, e);
            throw new IOException(msg, e);
        }
    }

    /**
     * Close batchwriter.
     * Throws a IOException as required by the flushable interface,
     * wrapping MutationsRejectedException.
     */
    @Override
    public void close() throws IOException {
        try {
            if (mtbw != null) {
                mtbw.close();
            }

        } catch (final MutationsRejectedException e) {
            final String msg = "Error while closing the batch writer.";
            logger.error(msg, e);
            throw new IOException(msg, e);
        }
    }


    @Override
    public String getTableName() {
       return makeTableName( ConfigUtils.getTablePrefix(conf) );
    }

    /**
     * Make the Accumulo table name used by this indexer for a specific instance of Rya.
     *
     * @param ryaInstanceName -  The name of the Rya instance the table name is for. (not null)
     * @return The Accumulo table name used by this indexer for a specific instance of Rya.
     */
    public static String makeTableName(final String ryaInstanceName) {
        requireNonNull(ryaInstanceName);
        return ryaInstanceName + TABLE_SUFFIX;
    }

    private void deleteStatement(final Statement statement) throws IOException, IllegalArgumentException {
    	Objects.requireNonNull(temporalIndexBatchWriter,"This is not initialized for writing.  Must call setMultiTableBatchWriter() and init().");

        // if the predicate list is empty, accept all predicates.
        // Otherwise, make sure the predicate is on the "valid" list
        final boolean isValidPredicate = validPredicates.isEmpty() || validPredicates.contains(statement.getPredicate());
        if (!isValidPredicate || !(statement.getObject() instanceof Literal)) {
            return;
        }
        final DateTime[] indexDateTimes = new DateTime[2]; // 0 begin, 1 end of interval
        extractDateTime(statement, indexDateTimes);
        if (indexDateTimes[0] == null) {
            return;
        }

        // Remove this as an instant, or interval.
        try {
            if (indexDateTimes[1] != null) {
                final TemporalInterval interval = new TemporalInterval(new TemporalInstantRfc3339(indexDateTimes[0]), new TemporalInstantRfc3339(indexDateTimes[1]));
                removeInterval(temporalIndexBatchWriter, interval, statement);
            } else {
                final TemporalInstant instant = new TemporalInstantRfc3339(indexDateTimes[0]);
                removeInstant(temporalIndexBatchWriter, instant, statement);
            }
        } catch (final MutationsRejectedException e) {
            throw new IOException("While adding interval/instant for statement =" + statement, e);
        }
    }

    @Override
    public void deleteStatement(final RyaStatement statement) throws IllegalArgumentException, IOException {
        deleteStatement(RyaToRdfConversions.convertStatement(statement));
    }

    @Override
    public void setConnector(final Connector connector) {
        // TODO Auto-generated method stub

    }

    @Override
    public void destroy() {
        // TODO Auto-generated method stub

    }

    @Override
    public void purge(final RdfCloudTripleStoreConfiguration configuration) {
        // TODO Auto-generated method stub

    }

    @Override
    public void dropAndDestroy() {
        // TODO Auto-generated method stub

    }

    /*
     * (non-Javadoc)
     * 
     * @see org.apache.rya.accumulo.experimental.AbstractAccumuloIndexer#setMultiTableBatchWriter(org.apache.accumulo.core.client.MultiTableBatchWriter)
     */
    @Override
    public void setMultiTableBatchWriter(MultiTableBatchWriter writer) throws IOException {
        mtbw = writer;
    }
}