/*
 * 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.pcj.fluo.integration;

import static java.util.Objects.requireNonNull;
import static org.junit.Assert.assertEquals;

import java.math.BigDecimal;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Collection;
import java.util.HashSet;
import java.util.Optional;
import java.util.Set;

import javax.xml.datatype.DatatypeFactory;

import org.apache.accumulo.core.client.Connector;
import org.apache.fluo.api.client.FluoClient;
import org.apache.fluo.core.client.FluoClientImpl;
import org.apache.rya.api.client.CreatePCJ.ExportStrategy;
import org.apache.rya.api.client.RyaClient;
import org.apache.rya.api.client.accumulo.AccumuloRyaClientFactory;
import org.apache.rya.api.functions.DateTimeWithinPeriod;
import org.apache.rya.api.functions.OWLTime;
import org.apache.rya.api.utils.CloseableIterator;
import org.apache.rya.indexing.pcj.fluo.api.CreateFluoPcj;
import org.apache.rya.indexing.pcj.fluo.app.query.UnsupportedQueryException;
import org.apache.rya.indexing.pcj.storage.PeriodicQueryResultStorage;
import org.apache.rya.indexing.pcj.storage.PrecomputedJoinStorage;
import org.apache.rya.indexing.pcj.storage.accumulo.AccumuloPcjStorage;
import org.apache.rya.indexing.pcj.storage.accumulo.AccumuloPeriodicQueryResultStorage;
import org.apache.rya.pcj.fluo.test.base.RyaExportITBase;
import org.junit.Test;
import org.openrdf.model.Literal;
import org.openrdf.model.Statement;
import org.openrdf.model.URI;
import org.openrdf.model.Value;
import org.openrdf.model.ValueFactory;
import org.openrdf.model.datatypes.XMLDatatypeUtil;
import org.openrdf.model.impl.BooleanLiteralImpl;
import org.openrdf.model.impl.ValueFactoryImpl;
import org.openrdf.model.vocabulary.FN;
import org.openrdf.model.vocabulary.XMLSchema;
import org.openrdf.query.BindingSet;
import org.openrdf.query.algebra.evaluation.ValueExprEvaluationException;
import org.openrdf.query.algebra.evaluation.function.Function;
import org.openrdf.query.algebra.evaluation.function.FunctionRegistry;
import org.openrdf.query.impl.MapBindingSet;
import org.openrdf.repository.RepositoryException;
import org.openrdf.repository.sail.SailRepositoryConnection;

import com.google.common.collect.Sets;

/**
 * Performs integration tests over the Fluo application geared towards various query structures.
 */
public class QueryIT extends RyaExportITBase {

    @Test
    public void optionalStatements() throws Exception {
        // A query that has optional statement patterns. This query is looking for all
        // people who have Law degrees and any BAR exams they have passed (though they
        // do not have to have passed any).
        final String sparql = "SELECT ?person ?exam " + "WHERE {" + "?person <http://hasDegreeIn> <http://Law> . "
                + "OPTIONAL {?person <http://passedExam> ?exam } . " + "}";

        // Create the Statements that will be loaded into Rya.
        final ValueFactory vf = new ValueFactoryImpl();
        final Collection<Statement> statements = Sets.newHashSet(
                vf.createStatement(vf.createURI("http://Alice"), vf.createURI("http://hasDegreeIn"),
                        vf.createURI("http://Computer Science")),
                vf.createStatement(vf.createURI("http://Alice"), vf.createURI("http://passedExam"),
                        vf.createURI("http://Certified Ethical Hacker")),
                vf.createStatement(vf.createURI("http://Bob"), vf.createURI("http://hasDegreeIn"), vf.createURI("http://Law")),
                vf.createStatement(vf.createURI("http://Bob"), vf.createURI("http://passedExam"), vf.createURI("http://MBE")),
                vf.createStatement(vf.createURI("http://Bob"), vf.createURI("http://passedExam"), vf.createURI("http://BAR-Kansas")),
                vf.createStatement(vf.createURI("http://Charlie"), vf.createURI("http://hasDegreeIn"), vf.createURI("http://Law")));

        // Create the expected results of the SPARQL query once the PCJ has been computed.
        final Set<BindingSet> expectedResults = new HashSet<>();

        MapBindingSet bs = new MapBindingSet();
        bs.addBinding("person", vf.createURI("http://Bob"));
        bs.addBinding("exam", vf.createURI("http://MBE"));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("person", vf.createURI("http://Bob"));
        bs.addBinding("exam", vf.createURI("http://BAR-Kansas"));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("person", vf.createURI("http://Charlie"));
        expectedResults.add(bs);

        // Verify the end results of the query match the expected results.
        runTest(sparql, statements, expectedResults, ExportStrategy.RYA);
    }

    /**
     * Tests when there are a bunch of variables across a bunch of joins.
     */
    @Test
    public void complexQuery() throws Exception {
        // A query that find people who live in the USA, have been recruited by Geek Squad,
        // and are skilled with computers. The resulting binding set includes everybody who
        // was involved in the recruitment process.
        final String sparql = "SELECT ?recruiter ?candidate ?leader " + "{ " + "?recruiter <http://recruiterFor> <http://GeekSquad>. "
                + "?recruiter <http://talksTo> ?candidate. " + "?candidate <http://skilledWith> <http://Computers>. " + "?candidate <http://livesIn> \"USA\". "
                + "?candidate <http://talksTo> ?leader." + "?leader <http://leaderOf> <http://GeekSquad>. }";

        // Create the Statements that will be loaded into Rya.
        final ValueFactory vf = new ValueFactoryImpl();
        final Collection<Statement> statements = Sets.newHashSet(
                // Leaders
                vf.createStatement(vf.createURI("http://Alice"), vf.createURI("http://leaderOf"), vf.createURI("http://GeekSquad")),
                vf.createStatement(vf.createURI("http://Bob"), vf.createURI("http://leaderOf"), vf.createURI("http://GeekSquad")),

        // Recruiters
                vf.createStatement(vf.createURI("http://Charlie"), vf.createURI("http://recruiterFor"), vf.createURI("http://GeekSquad")),
                vf.createStatement(vf.createURI("http://David"), vf.createURI("http://recruiterFor"), vf.createURI("http://GeekSquad")),

        // Candidates
                vf.createStatement(vf.createURI("http://Eve"), vf.createURI("http://skilledWith"), vf.createURI("http://Computers")),
                vf.createStatement(vf.createURI("http://Eve"), vf.createURI("http://livesIn"), vf.createLiteral("USA")),
                vf.createStatement(vf.createURI("http://Frank"), vf.createURI("http://skilledWith"), vf.createURI("http://Computers")),
                vf.createStatement(vf.createURI("http://Frank"), vf.createURI("http://livesIn"), vf.createLiteral("USA")),
                vf.createStatement(vf.createURI("http://George"), vf.createURI("http://skilledWith"), vf.createURI("http://Computers")),
                vf.createStatement(vf.createURI("http://George"), vf.createURI("http://livesIn"), vf.createLiteral("Germany")),
                vf.createStatement(vf.createURI("http://Harry"), vf.createURI("http://skilledWith"), vf.createURI("http://Negotiating")),
                vf.createStatement(vf.createURI("http://Harry"), vf.createURI("http://livesIn"), vf.createLiteral("USA")),
                vf.createStatement(vf.createURI("http://Ivan"), vf.createURI("http://skilledWith"), vf.createURI("http://Computers")),
                vf.createStatement(vf.createURI("http://Ivan"), vf.createURI("http://livesIn"), vf.createLiteral("USA")),

        // Candidates the recruiters talk to.
                vf.createStatement(vf.createURI("http://Charlie"), vf.createURI("http://talksTo"), vf.createURI("http://Eve")),
                vf.createStatement(vf.createURI("http://Charlie"), vf.createURI("http://talksTo"), vf.createURI("http://George")),
                vf.createStatement(vf.createURI("http://Charlie"), vf.createURI("http://talksTo"), vf.createURI("http://Harry")),
                vf.createStatement(vf.createURI("http://David"), vf.createURI("http://talksTo"), vf.createURI("http://Eve")),
                vf.createStatement(vf.createURI("http://David"), vf.createURI("http://talksTo"), vf.createURI("http://Frank")),
                vf.createStatement(vf.createURI("http://David"), vf.createURI("http://talksTo"), vf.createURI("http://Ivan")),

        // Recruits that talk to leaders.
                vf.createStatement(vf.createURI("http://Eve"), vf.createURI("http://talksTo"), vf.createURI("http://Alice")),
                vf.createStatement(vf.createURI("http://George"), vf.createURI("http://talksTo"), vf.createURI("http://Alice")),
                vf.createStatement(vf.createURI("http://Harry"), vf.createURI("http://talksTo"), vf.createURI("http://Bob")),
                vf.createStatement(vf.createURI("http://Ivan"), vf.createURI("http://talksTo"), vf.createURI("http://Bob")));

        // Create the expected results of the SPARQL query once the PCJ has been computed.
        final Set<BindingSet> expectedResults = new HashSet<>();

        MapBindingSet bs = new MapBindingSet();
        bs.addBinding("recruiter", vf.createURI("http://Charlie"));
        bs.addBinding("candidate", vf.createURI("http://Eve"));
        bs.addBinding("leader", vf.createURI("http://Alice"));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("recruiter", vf.createURI("http://David"));
        bs.addBinding("candidate", vf.createURI("http://Eve"));
        bs.addBinding("leader", vf.createURI("http://Alice"));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("recruiter", vf.createURI("http://David"));
        bs.addBinding("candidate", vf.createURI("http://Ivan"));
        bs.addBinding("leader", vf.createURI("http://Bob"));
        expectedResults.add(bs);

        // Verify the end results of the query match the expected results.
        runTest(sparql, statements, expectedResults, ExportStrategy.RYA);
    }

    @Test
    public void withURIFilters() throws Exception {
        final String sparql = "SELECT ?customer ?worker ?city " + "{ " + "FILTER(?customer = <http://Alice>) "
                + "FILTER(?city = <http://London>) " + "?customer <http://talksTo> ?worker. " + "?worker <http://livesIn> ?city. "
                + "?worker <http://worksAt> <http://Chipotle>. " + "}";

        // Create the Statements that will be loaded into Rya.
        final ValueFactory vf = new ValueFactoryImpl();
        final Collection<Statement> statements = Sets.newHashSet(
                vf.createStatement(vf.createURI("http://Alice"), vf.createURI("http://talksTo"), vf.createURI("http://Bob")),
                vf.createStatement(vf.createURI("http://Bob"), vf.createURI("http://livesIn"), vf.createURI("http://London")),
                vf.createStatement(vf.createURI("http://Bob"), vf.createURI("http://worksAt"), vf.createURI("http://Chipotle")),

        vf.createStatement(vf.createURI("http://Alice"), vf.createURI("http://talksTo"), vf.createURI("http://Charlie")),
                vf.createStatement(vf.createURI("http://Charlie"), vf.createURI("http://livesIn"), vf.createURI("http://London")),
                vf.createStatement(vf.createURI("http://Charlie"), vf.createURI("http://worksAt"), vf.createURI("http://Chipotle")),

        vf.createStatement(vf.createURI("http://Alice"), vf.createURI("http://talksTo"), vf.createURI("http://David")),
                vf.createStatement(vf.createURI("http://David"), vf.createURI("http://livesIn"), vf.createURI("http://London")),
                vf.createStatement(vf.createURI("http://David"), vf.createURI("http://worksAt"), vf.createURI("http://Chipotle")),

        vf.createStatement(vf.createURI("http://Alice"), vf.createURI("http://talksTo"), vf.createURI("http://Eve")),
                vf.createStatement(vf.createURI("http://Eve"), vf.createURI("http://livesIn"), vf.createURI("http://Leeds")),
                vf.createStatement(vf.createURI("http://Eve"), vf.createURI("http://worksAt"), vf.createURI("http://Chipotle")),

        vf.createStatement(vf.createURI("http://Frank"), vf.createURI("http://talksTo"), vf.createURI("http://Alice")),
                vf.createStatement(vf.createURI("http://Frank"), vf.createURI("http://livesIn"), vf.createURI("http://London")),
                vf.createStatement(vf.createURI("http://Frank"), vf.createURI("http://worksAt"), vf.createURI("http://Chipotle")));

        // Create the expected results of the SPARQL query once the PCJ has been computed.
        final Set<BindingSet> expectedResults = new HashSet<>();

        MapBindingSet bs = new MapBindingSet();
        bs.addBinding("customer", vf.createURI("http://Alice"));
        bs.addBinding("worker", vf.createURI("http://Bob"));
        bs.addBinding("city", vf.createURI("http://London"));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("customer", vf.createURI("http://Alice"));
        bs.addBinding("worker", vf.createURI("http://Charlie"));
        bs.addBinding("city", vf.createURI("http://London"));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("customer", vf.createURI("http://Alice"));
        bs.addBinding("worker", vf.createURI("http://David"));
        bs.addBinding("city", vf.createURI("http://London"));
        expectedResults.add(bs);

        // Verify the end results of the query match the expected results.
        runTest(sparql, statements, expectedResults, ExportStrategy.RYA);
    }

    @Test
    public void withNumericFilters() throws Exception {
        final String sparql = "SELECT ?name ?age " + "{" + "FILTER(?age < 30) ." + "?name <http://hasAge> ?age."
                + "?name <http://playsSport> \"Soccer\" " + "}";

        // Create the Statements that will be loaded into Rya.
        final ValueFactory vf = new ValueFactoryImpl();
        final Collection<Statement> statements = Sets.newHashSet(
                vf.createStatement(vf.createURI("http://Alice"), vf.createURI("http://hasAge"), vf.createLiteral(18)),
                vf.createStatement(vf.createURI("http://Bob"), vf.createURI("http://hasAge"), vf.createLiteral(30)),
                vf.createStatement(vf.createURI("http://Charlie"), vf.createURI("http://hasAge"), vf.createLiteral(14)),
                vf.createStatement(vf.createURI("http://David"), vf.createURI("http://hasAge"), vf.createLiteral(16)),
                vf.createStatement(vf.createURI("http://Eve"), vf.createURI("http://hasAge"), vf.createLiteral(35)),

        vf.createStatement(vf.createURI("http://Alice"), vf.createURI("http://playsSport"), vf.createLiteral("Soccer")),
                vf.createStatement(vf.createURI("http://Bob"), vf.createURI("http://playsSport"), vf.createLiteral("Soccer")),
                vf.createStatement(vf.createURI("http://Charlie"), vf.createURI("http://playsSport"), vf.createLiteral("Basketball")),
                vf.createStatement(vf.createURI("http://Charlie"), vf.createURI("http://playsSport"), vf.createLiteral("Soccer")),
                vf.createStatement(vf.createURI("http://David"), vf.createURI("http://playsSport"), vf.createLiteral("Basketball")));

        // Create the expected results of the SPARQL query once the PCJ has been computed.
        final Set<BindingSet> expectedResults = new HashSet<>();

        MapBindingSet bs = new MapBindingSet();
        bs.addBinding("name", vf.createURI("http://Alice"));
        bs.addBinding("age", vf.createLiteral("18", XMLSchema.INTEGER));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("name", vf.createURI("http://Charlie"));
        bs.addBinding("age", vf.createLiteral("14", XMLSchema.INTEGER));
        expectedResults.add(bs);

        // Verify the end results of the query match the expected results.
        runTest(sparql, statements, expectedResults, ExportStrategy.RYA);
    }

    @Test
    public void withCustomFilters() throws Exception {
        final String sparql = "prefix ryafunc: <tag:rya.apache.org,2017:function#> " + "SELECT ?name ?age " + "{ "
                + "FILTER( ryafunc:isTeen(?age) ) . " + "?name <http://hasAge> ?age . " + "?name <http://playsSport> \"Soccer\" . " + "}";

        // Register a custom Filter.
        final Function fooFunction = new Function() {
            @Override
            public String getURI() {
                return "tag:rya.apache.org,2017:function#isTeen";
            }

            final static int TEEN_THRESHOLD = 20;

            @Override
            public Value evaluate(final ValueFactory valueFactory, final Value... args) throws ValueExprEvaluationException {
                if (args.length != 1) {
                    throw new ValueExprEvaluationException("isTeen() requires exactly 1 argument, got " + args.length);
                }

                if (args[0] instanceof Literal) {
                    final Literal literal = (Literal) args[0];
                    final URI datatype = literal.getDatatype();

                    // ABS function accepts only numeric literals
                    if (datatype != null && XMLDatatypeUtil.isNumericDatatype(datatype)) {
                        if (XMLDatatypeUtil.isDecimalDatatype(datatype)) {
                            final BigDecimal bigValue = literal.decimalValue();
                            return BooleanLiteralImpl.valueOf(bigValue.compareTo(new BigDecimal(TEEN_THRESHOLD)) < 0);
                        } else if (XMLDatatypeUtil.isFloatingPointDatatype(datatype)) {
                            final double doubleValue = literal.doubleValue();
                            return BooleanLiteralImpl.valueOf(doubleValue < TEEN_THRESHOLD);
                        } else {
                            throw new ValueExprEvaluationException(
                                    "unexpected datatype (expect decimal/int or floating) for function operand: " + args[0]);
                        }
                    } else {
                        throw new ValueExprEvaluationException(
                                "unexpected input value (expect non-null and numeric) for function: " + args[0]);
                    }
                } else {
                    throw new ValueExprEvaluationException("unexpected input value (expect literal) for function: " + args[0]);
                }
            }
        };

        // Add our new function to the registry
        FunctionRegistry.getInstance().add(fooFunction);

        // Create the Statements that will be loaded into Rya.
        final ValueFactory vf = new ValueFactoryImpl();
        final Collection<Statement> statements = Sets.newHashSet(
                vf.createStatement(vf.createURI("http://Alice"), vf.createURI("http://hasAge"), vf.createLiteral(18)),
                vf.createStatement(vf.createURI("http://Bob"), vf.createURI("http://hasAge"), vf.createLiteral(30)),
                vf.createStatement(vf.createURI("http://Charlie"), vf.createURI("http://hasAge"), vf.createLiteral(14)),
                vf.createStatement(vf.createURI("http://David"), vf.createURI("http://hasAge"), vf.createLiteral(16)),
                vf.createStatement(vf.createURI("http://Eve"), vf.createURI("http://hasAge"), vf.createLiteral(35)),

        vf.createStatement(vf.createURI("http://Alice"), vf.createURI("http://playsSport"), vf.createLiteral("Soccer")),
                vf.createStatement(vf.createURI("http://Bob"), vf.createURI("http://playsSport"), vf.createLiteral("Soccer")),
                vf.createStatement(vf.createURI("http://Charlie"), vf.createURI("http://playsSport"), vf.createLiteral("Basketball")),
                vf.createStatement(vf.createURI("http://Charlie"), vf.createURI("http://playsSport"), vf.createLiteral("Soccer")),
                vf.createStatement(vf.createURI("http://David"), vf.createURI("http://playsSport"), vf.createLiteral("Basketball")));

        // Create the expected results of the SPARQL query once the PCJ has been computed.
        final Set<BindingSet> expectedResults = new HashSet<>();

        MapBindingSet bs = new MapBindingSet();
        bs.addBinding("name", vf.createURI("http://Alice"));
        bs.addBinding("age", vf.createLiteral("18", XMLSchema.INTEGER));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("name", vf.createURI("http://Charlie"));
        bs.addBinding("age", vf.createLiteral("14", XMLSchema.INTEGER));
        expectedResults.add(bs);

        // Verify the end results of the query match the expected results.
        runTest(sparql, statements, expectedResults, ExportStrategy.RYA);
    }

    @Test
    public void withTemporal() throws Exception {
        // A query that finds all stored data after 3 seconds.
        final String dtPredUri = "http://www.w3.org/2006/time#inXSDDateTime";
        final String dtPred = "<" + dtPredUri + ">";

        final String sparql = "PREFIX time: <http://www.w3.org/2006/time#> " + "PREFIX xml: <http://www.w3.org/2001/XMLSchema#> "
                + "PREFIX tempo: <tag:rya-rdf.org,2015:temporal#> " + "SELECT ?event ?time " + "WHERE { " + "?event " + dtPred + " ?time . "
                + "FILTER(?time > '2001-01-01T01:01:03-08:00'^^xml:dateTime) " + "}";

        // Create the Statements that will be loaded into Rya.
        final ValueFactory vf = new ValueFactoryImpl();
        final DatatypeFactory dtf = DatatypeFactory.newInstance();
        final Collection<Statement> statements = Sets.newHashSet(
                vf.createStatement(vf.createURI("http://eventz"), vf.createURI("http://www.w3.org/1999/02/22-rdf-syntax-ns#type"),
                        vf.createURI("http://www.w3.org/2006/time#Instant")),
                vf.createStatement(vf.createURI("http://eventz"), vf.createURI(dtPredUri),
                        vf.createLiteral(dtf.newXMLGregorianCalendar("2001-01-01T01:01:01-08:00"))), // 1 second
                vf.createStatement(vf.createURI("http://eventz"), vf.createURI(dtPredUri),
                        vf.createLiteral(dtf.newXMLGregorianCalendar("2001-01-01T04:01:02.000-05:00"))), // 2 second
                vf.createStatement(vf.createURI("http://eventz"), vf.createURI(dtPredUri),
                        vf.createLiteral(dtf.newXMLGregorianCalendar("2001-01-01T01:01:03-08:00"))), // 3 seconds
                vf.createStatement(vf.createURI("http://eventz"), vf.createURI(dtPredUri),
                        vf.createLiteral(dtf.newXMLGregorianCalendar("2001-01-01T01:01:04-08:00"))), // 4 seconds
                vf.createStatement(vf.createURI("http://eventz"), vf.createURI(dtPredUri),
                        vf.createLiteral(dtf.newXMLGregorianCalendar("2001-01-01T09:01:05Z"))), // 5 seconds
                vf.createStatement(vf.createURI("http://eventz"), vf.createURI(dtPredUri),
                        vf.createLiteral(dtf.newXMLGregorianCalendar("2006-01-01T05:00:00.000Z"))),
                vf.createStatement(vf.createURI("http://eventz"), vf.createURI(dtPredUri),
                        vf.createLiteral(dtf.newXMLGregorianCalendar("2007-01-01T05:00:00.000Z"))),
                vf.createStatement(vf.createURI("http://eventz"), vf.createURI(dtPredUri),
                        vf.createLiteral(dtf.newXMLGregorianCalendar("2008-01-01T05:00:00.000Z"))));

        // Create the expected results of the SPARQL query once the PCJ has been computed.
        final Set<BindingSet> expectedResults = new HashSet<>();

        MapBindingSet bs = new MapBindingSet();
        bs.addBinding("event", vf.createURI("http://eventz"));
        bs.addBinding("time", vf.createLiteral(dtf.newXMLGregorianCalendar("2001-01-01T09:01:04.000Z")));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("event", vf.createURI("http://eventz"));
        bs.addBinding("time", vf.createLiteral(dtf.newXMLGregorianCalendar("2001-01-01T09:01:05.000Z")));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("event", vf.createURI("http://eventz"));
        bs.addBinding("time", vf.createLiteral(dtf.newXMLGregorianCalendar("2006-01-01T05:00:00.000Z")));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("event", vf.createURI("http://eventz"));
        bs.addBinding("time", vf.createLiteral(dtf.newXMLGregorianCalendar("2007-01-01T05:00:00.000Z")));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("event", vf.createURI("http://eventz"));
        bs.addBinding("time", vf.createLiteral(dtf.newXMLGregorianCalendar("2008-01-01T05:00:00.000Z")));
        expectedResults.add(bs);

        // Verify the end results of the query match the expected results.
        runTest(sparql, statements, expectedResults, ExportStrategy.RYA);
    }


    @Test
    public void dateTimeWithin() throws Exception {

        final ValueFactory vf = new ValueFactoryImpl();
        final DatatypeFactory dtf = DatatypeFactory.newInstance();
        FunctionRegistry.getInstance().add(new DateTimeWithinPeriod());

        final String sparql = "PREFIX fn: <" + FN.NAMESPACE +">"
                + "SELECT ?event ?startTime ?endTime WHERE { ?event <uri:startTime> ?startTime; <uri:endTime> ?endTime. "
                + "FILTER(fn:dateTimeWithin(?startTime, ?endTime, 2,<" + OWLTime.HOURS_URI + "> ))}";

        final ZonedDateTime zTime = ZonedDateTime.now();
        final String time = zTime.format(DateTimeFormatter.ISO_INSTANT);

        final ZonedDateTime zTime1 = zTime.minusHours(1);
        final String time1 = zTime1.format(DateTimeFormatter.ISO_INSTANT);

        final ZonedDateTime zTime2 = zTime.minusHours(2);
        final String time2 = zTime2.format(DateTimeFormatter.ISO_INSTANT);

        final Literal lit = vf.createLiteral(dtf.newXMLGregorianCalendar(time));
        final Literal lit1 = vf.createLiteral(dtf.newXMLGregorianCalendar(time1));
        final Literal lit2 = vf.createLiteral(dtf.newXMLGregorianCalendar(time2));

        // Create the Statements that will be loaded into Rya.
        final Collection<Statement> statements = Sets.newHashSet(
                vf.createStatement(vf.createURI("uri:event1"), vf.createURI("uri:startTime"), lit),
                vf.createStatement(vf.createURI("uri:event1"), vf.createURI("uri:endTime"), lit1),
                vf.createStatement(vf.createURI("uri:event2"), vf.createURI("uri:startTime"), lit),
                vf.createStatement(vf.createURI("uri:event2"), vf.createURI("uri:endTime"), lit2)
               );

        // Create the expected results of the SPARQL query once the PCJ has been computed.
        final Set<BindingSet> expectedResults = new HashSet<>();

        final MapBindingSet bs = new MapBindingSet();
        bs.addBinding("event", vf.createURI("uri:event1"));
        bs.addBinding("startTime", lit);
        bs.addBinding("endTime", lit1);
        expectedResults.add(bs);

        // Verify the end results of the query match the expected results.
        runTest(sparql, statements, expectedResults, ExportStrategy.RYA);
    }

    @Test
    public void dateTimeWithinNow() throws Exception {

        final ValueFactory vf = new ValueFactoryImpl();
        final DatatypeFactory dtf = DatatypeFactory.newInstance();
        FunctionRegistry.getInstance().add(new DateTimeWithinPeriod());

        final String sparql = "PREFIX fn: <" + FN.NAMESPACE +">"
                + "SELECT ?event ?startTime WHERE { ?event <uri:startTime> ?startTime. "
                + "FILTER(fn:dateTimeWithin(?startTime, NOW(), 30, <" + OWLTime.SECONDS_URI + "> ))}";

        final ZonedDateTime zTime = ZonedDateTime.now();
        final String time = zTime.format(DateTimeFormatter.ISO_INSTANT);

        final ZonedDateTime zTime1 = zTime.minusSeconds(30);
        final String time1 = zTime1.format(DateTimeFormatter.ISO_INSTANT);

        final Literal lit = vf.createLiteral(dtf.newXMLGregorianCalendar(time));
        final Literal lit1 = vf.createLiteral(dtf.newXMLGregorianCalendar(time1));

        // Create the Statements that will be loaded into Rya.
        final Collection<Statement> statements = Sets.newHashSet(
                vf.createStatement(vf.createURI("uri:event1"), vf.createURI("uri:startTime"), lit),
                vf.createStatement(vf.createURI("uri:event2"), vf.createURI("uri:startTime"), lit1)
               );

        // Create the expected results of the SPARQL query once the PCJ has been computed.
        final Set<BindingSet> expectedResults = new HashSet<>();

        final MapBindingSet bs = new MapBindingSet();
        bs.addBinding("event", vf.createURI("uri:event1"));
        bs.addBinding("startTime", lit);
        expectedResults.add(bs);

        // Verify the end results of the query match the expected results.
        runTest(sparql, statements, expectedResults, ExportStrategy.RYA);
    }


    @Test
    public void periodicQueryTestWithoutAggregation() throws Exception {
        final String query = "prefix function: <http://org.apache.rya/function#> " // n
                + "prefix time: <http://www.w3.org/2006/time#> " // n
                + "select ?id where {" // n
                + "Filter(function:periodic(?time, 2, .5, time:hours)) " // n
                + "?obs <uri:hasTime> ?time. " // n
                + "?obs <uri:hasId> ?id }"; // n

        // Create the Statements that will be loaded into Rya.
        final ValueFactory vf = new ValueFactoryImpl();
        final DatatypeFactory dtf = DatatypeFactory.newInstance();
        final ZonedDateTime time = ZonedDateTime.now();
        final long currentTime = time.toInstant().toEpochMilli();

        final ZonedDateTime zTime1 = time.minusMinutes(30);
        final String time1 = zTime1.format(DateTimeFormatter.ISO_INSTANT);

        final ZonedDateTime zTime2 = zTime1.minusMinutes(30);
        final String time2 = zTime2.format(DateTimeFormatter.ISO_INSTANT);

        final ZonedDateTime zTime3 = zTime2.minusMinutes(30);
        final String time3 = zTime3.format(DateTimeFormatter.ISO_INSTANT);

        final ZonedDateTime zTime4 = zTime3.minusMinutes(30);
        final String time4 = zTime4.format(DateTimeFormatter.ISO_INSTANT);

        final Collection<Statement> statements = Sets.newHashSet(
                vf.createStatement(vf.createURI("urn:obs_1"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time1))),
                vf.createStatement(vf.createURI("urn:obs_1"), vf.createURI("uri:hasId"), vf.createLiteral("id_1")),
                vf.createStatement(vf.createURI("urn:obs_2"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time2))),
                vf.createStatement(vf.createURI("urn:obs_2"), vf.createURI("uri:hasId"), vf.createLiteral("id_2")),
                vf.createStatement(vf.createURI("urn:obs_3"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time3))),
                vf.createStatement(vf.createURI("urn:obs_3"), vf.createURI("uri:hasId"), vf.createLiteral("id_3")),
                vf.createStatement(vf.createURI("urn:obs_4"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time4))),
                vf.createStatement(vf.createURI("urn:obs_4"), vf.createURI("uri:hasId"), vf.createLiteral("id_4")));

        // Create the expected results of the SPARQL query once the PCJ has been computed.
        final Set<BindingSet> expectedResults = new HashSet<>();

        final long period = 1800000;
        final long binId = currentTime / period * period;

        MapBindingSet bs = new MapBindingSet();
        bs.addBinding("id", vf.createLiteral("id_1", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("id", vf.createLiteral("id_1", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId + period));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("id", vf.createLiteral("id_1", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId + 2 * period));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("id", vf.createLiteral("id_1", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId + 3 * period));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("id", vf.createLiteral("id_2", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("id", vf.createLiteral("id_2", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId + period));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("id", vf.createLiteral("id_2", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId + 2 * period));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("id", vf.createLiteral("id_3", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("id", vf.createLiteral("id_3", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId + period));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("id", vf.createLiteral("id_4", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId));
        expectedResults.add(bs);

        // Verify the end results of the query match the expected results.
        runTest(query, statements, expectedResults, ExportStrategy.PERIODIC);
    }

    @Test
    public void periodicQueryTestWithAggregation() throws Exception {
        final String query = "prefix function: <http://org.apache.rya/function#> " // n
                + "prefix time: <http://www.w3.org/2006/time#> " // n
                + "select (count(?obs) as ?total) where {" // n
                + "Filter(function:periodic(?time, 2, .5, time:hours)) " // n
                + "?obs <uri:hasTime> ?time. " // n
                + "?obs <uri:hasId> ?id }"; // n

        // Create the Statements that will be loaded into Rya.
        final ValueFactory vf = new ValueFactoryImpl();
        final DatatypeFactory dtf = DatatypeFactory.newInstance();
        final ZonedDateTime time = ZonedDateTime.now();
        final long currentTime = time.toInstant().toEpochMilli();

        final ZonedDateTime zTime1 = time.minusMinutes(30);
        final String time1 = zTime1.format(DateTimeFormatter.ISO_INSTANT);

        final ZonedDateTime zTime2 = zTime1.minusMinutes(30);
        final String time2 = zTime2.format(DateTimeFormatter.ISO_INSTANT);

        final ZonedDateTime zTime3 = zTime2.minusMinutes(30);
        final String time3 = zTime3.format(DateTimeFormatter.ISO_INSTANT);

        final ZonedDateTime zTime4 = zTime3.minusMinutes(30);
        final String time4 = zTime4.format(DateTimeFormatter.ISO_INSTANT);

        final Collection<Statement> statements = Sets.newHashSet(
                vf.createStatement(vf.createURI("urn:obs_1"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time1))),
                vf.createStatement(vf.createURI("urn:obs_1"), vf.createURI("uri:hasId"), vf.createLiteral("id_1")),
                vf.createStatement(vf.createURI("urn:obs_2"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time2))),
                vf.createStatement(vf.createURI("urn:obs_2"), vf.createURI("uri:hasId"), vf.createLiteral("id_2")),
                vf.createStatement(vf.createURI("urn:obs_3"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time3))),
                vf.createStatement(vf.createURI("urn:obs_3"), vf.createURI("uri:hasId"), vf.createLiteral("id_3")),
                vf.createStatement(vf.createURI("urn:obs_4"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time4))),
                vf.createStatement(vf.createURI("urn:obs_4"), vf.createURI("uri:hasId"), vf.createLiteral("id_4")));

        // Create the expected results of the SPARQL query once the PCJ has been computed.
        final Set<BindingSet> expectedResults = new HashSet<>();

        final long period = 1800000;
        final long binId = currentTime / period * period;

        MapBindingSet bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("4", XMLSchema.INTEGER));
        bs.addBinding("periodicBinId", vf.createLiteral(binId));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("3", XMLSchema.INTEGER));
        bs.addBinding("periodicBinId", vf.createLiteral(binId + period));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("2", XMLSchema.INTEGER));
        bs.addBinding("periodicBinId", vf.createLiteral(binId + 2 * period));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("1", XMLSchema.INTEGER));
        bs.addBinding("periodicBinId", vf.createLiteral(binId + 3 * period));
        expectedResults.add(bs);

        // Verify the end results of the query match the expected results.
        runTest(query, statements, expectedResults, ExportStrategy.PERIODIC);
    }

    @Test
    public void periodicQueryTestWithAggregationAndGroupBy() throws Exception {
        final String query = "prefix function: <http://org.apache.rya/function#> " // n
                + "prefix time: <http://www.w3.org/2006/time#> " // n
                + "select ?id (count(?obs) as ?total) where {" // n
                + "Filter(function:periodic(?time, 2, .5, time:hours)) " // n
                + "?obs <uri:hasTime> ?time. " // n
                + "?obs <uri:hasId> ?id } group by ?id"; // n

        // Create the Statements that will be loaded into Rya.
        final ValueFactory vf = new ValueFactoryImpl();
        final DatatypeFactory dtf = DatatypeFactory.newInstance();
        final ZonedDateTime time = ZonedDateTime.now();
        final long currentTime = time.toInstant().toEpochMilli();

        final ZonedDateTime zTime1 = time.minusMinutes(30);
        final String time1 = zTime1.format(DateTimeFormatter.ISO_INSTANT);

        final ZonedDateTime zTime2 = zTime1.minusMinutes(30);
        final String time2 = zTime2.format(DateTimeFormatter.ISO_INSTANT);

        final ZonedDateTime zTime3 = zTime2.minusMinutes(30);
        final String time3 = zTime3.format(DateTimeFormatter.ISO_INSTANT);

        final ZonedDateTime zTime4 = zTime3.minusMinutes(30);
        final String time4 = zTime4.format(DateTimeFormatter.ISO_INSTANT);

        final Collection<Statement> statements = Sets.newHashSet(
                vf.createStatement(vf.createURI("urn:obs_1"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time1))),
                vf.createStatement(vf.createURI("urn:obs_1"), vf.createURI("uri:hasId"), vf.createLiteral("id_1")),
                vf.createStatement(vf.createURI("urn:obs_2"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time2))),
                vf.createStatement(vf.createURI("urn:obs_2"), vf.createURI("uri:hasId"), vf.createLiteral("id_2")),
                vf.createStatement(vf.createURI("urn:obs_3"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time3))),
                vf.createStatement(vf.createURI("urn:obs_3"), vf.createURI("uri:hasId"), vf.createLiteral("id_3")),
                vf.createStatement(vf.createURI("urn:obs_4"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time4))),
                vf.createStatement(vf.createURI("urn:obs_4"), vf.createURI("uri:hasId"), vf.createLiteral("id_4")),
                vf.createStatement(vf.createURI("urn:obs_1"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time4))),
                vf.createStatement(vf.createURI("urn:obs_1"), vf.createURI("uri:hasId"), vf.createLiteral("id_1")),
                vf.createStatement(vf.createURI("urn:obs_2"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time3))),
                vf.createStatement(vf.createURI("urn:obs_2"), vf.createURI("uri:hasId"), vf.createLiteral("id_2")));

        // Create the expected results of the SPARQL query once the PCJ has been computed.
        final Set<BindingSet> expectedResults = new HashSet<>();

        final long period = 1800000;
        final long binId = currentTime / period * period;

        MapBindingSet bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("2", XMLSchema.INTEGER));
        bs.addBinding("id", vf.createLiteral("id_1", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("2", XMLSchema.INTEGER));
        bs.addBinding("id", vf.createLiteral("id_2", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("1", XMLSchema.INTEGER));
        bs.addBinding("id", vf.createLiteral("id_3", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("1", XMLSchema.INTEGER));
        bs.addBinding("id", vf.createLiteral("id_4", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("1", XMLSchema.INTEGER));
        bs.addBinding("id", vf.createLiteral("id_1", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId + period));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("2", XMLSchema.INTEGER));
        bs.addBinding("id", vf.createLiteral("id_2", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId + period));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("1", XMLSchema.INTEGER));
        bs.addBinding("id", vf.createLiteral("id_3", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId + period));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("1", XMLSchema.INTEGER));
        bs.addBinding("id", vf.createLiteral("id_1", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId + 2 * period));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("1", XMLSchema.INTEGER));
        bs.addBinding("id", vf.createLiteral("id_2", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId + 2 * period));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("1", XMLSchema.INTEGER));
        bs.addBinding("id", vf.createLiteral("id_1", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId + 3 * period));
        expectedResults.add(bs);

        // Verify the end results of the query match the expected results.
        runTest(query, statements, expectedResults, ExportStrategy.PERIODIC);
    }


    @Test
    public void nestedPeriodicQueryTestWithAggregationAndGroupBy() throws Exception {
        final String query = "prefix function: <http://org.apache.rya/function#> " // n
                + "prefix time: <http://www.w3.org/2006/time#> " // n
                + "select ?location ?total "
                + "where { Filter(?total > 1) {"
                + "select ?location (count(?obs) as ?total) where {" // n
                + "Filter(function:periodic(?time, 2, .5, time:hours)) " // n
                + "?obs <uri:hasTime> ?time. " // n
                + "?obs <uri:hasLoc> ?location } group by ?location }}"; // n

        // Create the Statements that will be loaded into Rya.
        final ValueFactory vf = new ValueFactoryImpl();
        final DatatypeFactory dtf = DatatypeFactory.newInstance();
        final ZonedDateTime time = ZonedDateTime.now();
        final long currentTime = time.toInstant().toEpochMilli();

        final ZonedDateTime zTime1 = time.minusMinutes(30);
        final String time1 = zTime1.format(DateTimeFormatter.ISO_INSTANT);

        final ZonedDateTime zTime2 = zTime1.minusMinutes(30);
        final String time2 = zTime2.format(DateTimeFormatter.ISO_INSTANT);

        final ZonedDateTime zTime3 = zTime2.minusMinutes(30);
        final String time3 = zTime3.format(DateTimeFormatter.ISO_INSTANT);

        final ZonedDateTime zTime4 = zTime3.minusMinutes(30);
        final String time4 = zTime4.format(DateTimeFormatter.ISO_INSTANT);

        final Collection<Statement> statements = Sets.newHashSet(
                vf.createStatement(vf.createURI("urn:obs_1"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time1))),
                vf.createStatement(vf.createURI("urn:obs_1"), vf.createURI("uri:hasLoc"), vf.createLiteral("loc_1")),
                vf.createStatement(vf.createURI("urn:obs_2"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time2))),
                vf.createStatement(vf.createURI("urn:obs_2"), vf.createURI("uri:hasLoc"), vf.createLiteral("loc_2")),
                vf.createStatement(vf.createURI("urn:obs_3"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time3))),
                vf.createStatement(vf.createURI("urn:obs_3"), vf.createURI("uri:hasLoc"), vf.createLiteral("loc_3")),
                vf.createStatement(vf.createURI("urn:obs_4"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time4))),
                vf.createStatement(vf.createURI("urn:obs_4"), vf.createURI("uri:hasLoc"), vf.createLiteral("loc_4")),
                vf.createStatement(vf.createURI("urn:obs_5"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time4))),
                vf.createStatement(vf.createURI("urn:obs_5"), vf.createURI("uri:hasLoc"), vf.createLiteral("loc_1")),
                vf.createStatement(vf.createURI("urn:obs_6"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time3))),
                vf.createStatement(vf.createURI("urn:obs_6"), vf.createURI("uri:hasLoc"), vf.createLiteral("loc_2")));

        // Create the expected results of the SPARQL query once the PCJ has been computed.
        final Set<BindingSet> expectedResults = new HashSet<>();

        final long period = 1800000;
        final long binId = currentTime / period * period;

        MapBindingSet bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("2", XMLSchema.INTEGER));
        bs.addBinding("location", vf.createLiteral("loc_1", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("2", XMLSchema.INTEGER));
        bs.addBinding("location", vf.createLiteral("loc_2", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId));
        expectedResults.add(bs);


        bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("2", XMLSchema.INTEGER));
        bs.addBinding("location", vf.createLiteral("loc_2", XMLSchema.STRING));
        bs.addBinding("periodicBinId", vf.createLiteral(binId + period));
        expectedResults.add(bs);

        // Verify the end results of the query match the expected results.
        runTest(query, statements, expectedResults, ExportStrategy.PERIODIC);
    }

    @Test
    public void nestedJoinPeriodicQueryWithAggregationAndGroupBy() throws Exception {
        final String query = "prefix function: <http://org.apache.rya/function#> " // n
                + "prefix time: <http://www.w3.org/2006/time#> " // n
                + "select ?location ?total ?population "
                + "where { Filter(?total > 1)"
                + "?location <uri:hasPopulation> ?population . {"
                + "select ?location (count(?obs) as ?total) where {" // n
                + "Filter(function:periodic(?time, 2, .5, time:hours)) " // n
                + "?obs <uri:hasTime> ?time. " // n
                + "?obs <uri:hasLoc> ?location } group by ?location }}"; // n

        // Create the Statements that will be loaded into Rya.
        final ValueFactory vf = new ValueFactoryImpl();
        final DatatypeFactory dtf = DatatypeFactory.newInstance();
        final ZonedDateTime time = ZonedDateTime.now();
        final long currentTime = time.toInstant().toEpochMilli();

        final ZonedDateTime zTime1 = time.minusMinutes(30);
        final String time1 = zTime1.format(DateTimeFormatter.ISO_INSTANT);

        final ZonedDateTime zTime2 = zTime1.minusMinutes(30);
        final String time2 = zTime2.format(DateTimeFormatter.ISO_INSTANT);

        final ZonedDateTime zTime3 = zTime2.minusMinutes(30);
        final String time3 = zTime3.format(DateTimeFormatter.ISO_INSTANT);

        final ZonedDateTime zTime4 = zTime3.minusMinutes(30);
        final String time4 = zTime4.format(DateTimeFormatter.ISO_INSTANT);

        final Collection<Statement> statements = Sets.newHashSet(
                vf.createStatement(vf.createURI("urn:obs_1"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time1))),
                vf.createStatement(vf.createURI("urn:obs_1"), vf.createURI("uri:hasLoc"), vf.createURI("uri:loc_1")),
                vf.createStatement(vf.createURI("uri:loc_1"), vf.createURI("uri:hasPopulation"), vf.createLiteral(3500)),
                vf.createStatement(vf.createURI("uri:loc_2"), vf.createURI("uri:hasPopulation"), vf.createLiteral(8000)),
                vf.createStatement(vf.createURI("urn:obs_2"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time2))),
                vf.createStatement(vf.createURI("urn:obs_2"), vf.createURI("uri:hasLoc"), vf.createURI("uri:loc_2")),
                vf.createStatement(vf.createURI("urn:obs_3"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time3))),
                vf.createStatement(vf.createURI("urn:obs_3"), vf.createURI("uri:hasLoc"), vf.createURI("uri:loc_3")),
                vf.createStatement(vf.createURI("urn:obs_4"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time4))),
                vf.createStatement(vf.createURI("urn:obs_4"), vf.createURI("uri:hasLoc"), vf.createURI("uri:loc_4")),
                vf.createStatement(vf.createURI("urn:obs_5"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time4))),
                vf.createStatement(vf.createURI("urn:obs_5"), vf.createURI("uri:hasLoc"), vf.createURI("uri:loc_1")),
                vf.createStatement(vf.createURI("urn:obs_6"), vf.createURI("uri:hasTime"),
                        vf.createLiteral(dtf.newXMLGregorianCalendar(time3))),
                vf.createStatement(vf.createURI("urn:obs_6"), vf.createURI("uri:hasLoc"), vf.createURI("uri:loc_2")));

        // Create the expected results of the SPARQL query once the PCJ has been computed.
        final Set<BindingSet> expectedResults = new HashSet<>();

        final long period = 1800000;
        final long binId = currentTime / period * period;

        MapBindingSet bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("2", XMLSchema.INTEGER));
        bs.addBinding("location", vf.createURI("uri:loc_1"));
        bs.addBinding("population", vf.createLiteral("3500", XMLSchema.INTEGER));
        bs.addBinding("periodicBinId", vf.createLiteral(binId));
        expectedResults.add(bs);

        bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("2", XMLSchema.INTEGER));
        bs.addBinding("location", vf.createURI("uri:loc_2"));
        bs.addBinding("population", vf.createLiteral("8000", XMLSchema.INTEGER));
        bs.addBinding("periodicBinId", vf.createLiteral(binId));
        expectedResults.add(bs);


        bs = new MapBindingSet();
        bs.addBinding("total", vf.createLiteral("2", XMLSchema.INTEGER));
        bs.addBinding("location", vf.createURI("uri:loc_2"));
        bs.addBinding("population", vf.createLiteral("8000", XMLSchema.INTEGER));
        bs.addBinding("periodicBinId", vf.createLiteral(binId + period));
        expectedResults.add(bs);

        // Verify the end results of the query match the expected results.
        runTest(query, statements, expectedResults, ExportStrategy.PERIODIC);
    }

    @Test(expected= UnsupportedQueryException.class)
    public void nestedConstructPeriodicQueryWithAggregationAndGroupBy() throws Exception {
        final String query = "prefix function: <http://org.apache.rya/function#> " // n
                + "prefix time: <http://www.w3.org/2006/time#> " // n
                + "construct{?location a <uri:highObservationArea> } "
                + "where { Filter(?total > 1)"
                + "?location <uri:hasPopulation> ?population . {"
                + "select ?location (count(?obs) as ?total) where {" // n
                + "Filter(function:periodic(?time, 2, .5, time:hours)) " // n
                + "?obs <uri:hasTime> ?time. " // n
                + "?obs <uri:hasLoc> ?location } group by ?location }}"; // n


        final Collection<Statement> statements = Sets.newHashSet();
        final Set<BindingSet> expectedResults = new HashSet<>();

        // Verify the end results of the query match the expected results.
        runTest(query, statements, expectedResults, ExportStrategy.PERIODIC);
    }

    public void runTest(final String sparql, final Collection<Statement> statements, final Collection<BindingSet> expectedResults,
            final ExportStrategy strategy) throws Exception {
        requireNonNull(sparql);
        requireNonNull(statements);
        requireNonNull(expectedResults);

        // Register the PCJ with Rya.
        final Connector accumuloConn = super.getAccumuloConnector();

        final RyaClient ryaClient = AccumuloRyaClientFactory.build(createConnectionDetails(), accumuloConn);

        switch (strategy) {
        case RYA:
            ryaClient.getCreatePCJ().createPCJ(getRyaInstanceName(), sparql);
            addStatementsAndWait(statements);
            // Fetch the value that is stored within the PCJ table.
            try (final PrecomputedJoinStorage pcjStorage = new AccumuloPcjStorage(accumuloConn, getRyaInstanceName())) {
                final String pcjId = pcjStorage.listPcjs().get(0);
                final Set<BindingSet> results = Sets.newHashSet(pcjStorage.listResults(pcjId));
                // Ensure the result of the query matches the expected result.
                assertEquals(expectedResults, results);
            }

            break;
        case PERIODIC:
            final PeriodicQueryResultStorage periodicStorage = new AccumuloPeriodicQueryResultStorage(accumuloConn, getRyaInstanceName());
            final String periodicId = periodicStorage.createPeriodicQuery(sparql);
            try (FluoClient fluo = new FluoClientImpl(super.getFluoConfiguration())) {
                new CreateFluoPcj().createPcj(periodicId, sparql, Sets.newHashSet(ExportStrategy.PERIODIC), fluo);
            }
            addStatementsAndWait(statements);

            final Set<BindingSet> results = Sets.newHashSet();
            try (CloseableIterator<BindingSet> resultIter = periodicStorage.listResults(periodicId, Optional.empty())) {
                while (resultIter.hasNext()) {
                    results.add(resultIter.next());
                }
            }
            assertEquals(expectedResults, results);
            break;
        default:
            throw new RuntimeException("Invalid export option");
        }
    }

    private void addStatementsAndWait(final Collection<Statement> statements) throws RepositoryException, Exception {
        // Write the data to Rya.
        final SailRepositoryConnection ryaConn = super.getRyaSailRepository().getConnection();
        ryaConn.begin();
        ryaConn.add(statements);
        ryaConn.commit();
        ryaConn.close();

        // Wait for the Fluo application to finish computing the end result.
        super.getMiniFluo().waitForObservers();
    }
}