You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rya.apache.org by mi...@apache.org on 2016/03/22 20:31:54 UTC
[3/7] incubator-rya git commit: RYA-55 Implemented a series of
projects that use Fluo to incrementally update the results of a Rya
Precomputed Join secondary index.
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/15ec5d5f/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/command/LoadTriplesCommand.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/command/LoadTriplesCommand.java b/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/command/LoadTriplesCommand.java
new file mode 100644
index 0000000..c24eb66
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/command/LoadTriplesCommand.java
@@ -0,0 +1,140 @@
+/*
+ * 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.client.command;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+import javax.annotation.ParametersAreNonnullByDefault;
+
+import org.apache.accumulo.core.client.Connector;
+import org.apache.commons.io.FilenameUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.rya.indexing.pcj.fluo.api.InsertTriples;
+import org.apache.rya.indexing.pcj.fluo.client.PcjAdminClientCommand;
+import org.apache.rya.indexing.pcj.fluo.client.util.FluoLoader;
+import org.openrdf.rio.RDFHandlerException;
+import org.openrdf.rio.RDFParseException;
+import org.openrdf.rio.RDFParser;
+import org.openrdf.rio.ntriples.NTriplesParser;
+import org.openrdf.rio.trig.TriGParser;
+import org.openrdf.rio.turtle.TurtleParser;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.ParameterException;
+
+import io.fluo.api.client.FluoClient;
+import mvm.rya.rdftriplestore.RyaSailRepository;
+
+/**
+ * A command that loads the contents of an NTriple file into the Fluo application.
+ */
+@ParametersAreNonnullByDefault
+public class LoadTriplesCommand implements PcjAdminClientCommand {
+ private static final Logger log = LogManager.getLogger(LoadTriplesCommand.class);
+
+ /**
+ * Command line parameters that are used by this command to configure itself.
+ */
+ private static final class Parameters {
+ @Parameter(names = "--triplesFile", required = true, description = "The RDF file of statemetns to load into the Fluo app.")
+ private String nTriplesFile;
+ }
+
+ @Override
+ public String getCommand() {
+ return "load-triples";
+ }
+
+ @Override
+ public String getDescription() {
+ return "Load RDF Triples into the Fluo app";
+ }
+
+ @Override
+ public String getUsage() {
+ final JCommander parser = new JCommander(new Parameters());
+
+ final StringBuilder usage = new StringBuilder();
+ parser.usage(usage);
+ return usage.toString();
+ }
+
+ @Override
+ public void execute(final Connector accumulo, final String ryaTablePrefix, final RyaSailRepository rya, final FluoClient fluo, final String[] args) throws ArgumentsException, ExecutionException {
+ checkNotNull(accumulo);
+ checkNotNull(fluo);
+ checkNotNull(args);
+
+ log.trace("Executing the Load Triples Command...");
+
+ // Parse the command line arguments.
+ final Parameters params = new Parameters();
+ try {
+ new JCommander(params, args);
+ } catch(final ParameterException e) {
+ throw new ArgumentsException("Could not load the Triples file because of invalid command line parameters.", e);
+ }
+
+ // Iterate over the Statements that are in the input file and write them to Fluo.
+ log.trace("Loading RDF Statements from the Triples file '" + params.nTriplesFile + "'.");
+ final Path triplesPath = Paths.get( params.nTriplesFile );
+
+ try {
+ final RDFParser parser = makeParser(triplesPath);
+ final FluoLoader loader = new FluoLoader(fluo, new InsertTriples());
+ parser.setRDFHandler(loader);
+ parser.parse(Files.newInputStream(triplesPath), triplesPath.toUri().toString());
+ } catch (UnsupportedFormatException | RDFParseException | RDFHandlerException | IOException e) {
+ throw new ExecutionException("Could not load the RDF file into the Fluo app.", e);
+ }
+
+ log.trace("Finished executing the Load Triples Command.");
+ }
+
+ private static RDFParser makeParser(final Path tripleFile) throws UnsupportedFormatException {
+ checkNotNull(tripleFile);
+
+ final String extension = FilenameUtils.getExtension( tripleFile.getFileName().toString() );
+ switch(extension) {
+ case "nt":
+ return new NTriplesParser();
+ case "ttl":
+ return new TurtleParser();
+ case "trig":
+ return new TriGParser();
+ default:
+ throw new UnsupportedFormatException("RDF File with extension '" + extension + "' not supported.");
+ }
+ }
+
+ private static final class UnsupportedFormatException extends Exception {
+ private static final long serialVersionUID = 1L;
+
+ public UnsupportedFormatException(final String message) {
+ super(message);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/15ec5d5f/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/command/NewQueryCommand.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/command/NewQueryCommand.java b/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/command/NewQueryCommand.java
new file mode 100644
index 0000000..1e15fc8
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/command/NewQueryCommand.java
@@ -0,0 +1,124 @@
+/*
+ * 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.client.command;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+import javax.annotation.ParametersAreNonnullByDefault;
+
+import org.apache.accumulo.core.client.Connector;
+import org.apache.commons.io.IOUtils;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.rya.indexing.pcj.fluo.api.CreatePcj;
+import org.apache.rya.indexing.pcj.fluo.client.PcjAdminClientCommand;
+import org.apache.rya.indexing.pcj.fluo.client.util.ParsedQueryRequest;
+import org.openrdf.query.MalformedQueryException;
+import org.openrdf.query.QueryEvaluationException;
+import org.openrdf.sail.SailException;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.ParameterException;
+
+import io.fluo.api.client.FluoClient;
+import mvm.rya.indexing.external.tupleSet.PcjTables.PcjException;
+import mvm.rya.rdftriplestore.RyaSailRepository;
+
+/**
+ * A command that creates a creates a new PCJ in the Fluo app and loads historic
+ * statement pattern matches for it.
+ */
+@ParametersAreNonnullByDefault
+public class NewQueryCommand implements PcjAdminClientCommand {
+ private static final Logger log = LogManager.getLogger(NewQueryCommand.class);
+
+ /**
+ * Command line parameters that are used by this command to configure itself.
+ */
+ private static final class Parameters {
+ @Parameter(names = "--queryRequestFile", required = true, description = "The path to a file containing the SPARQL query that will be loaded into the Fluo app.")
+ private String queryRequestFile;
+ }
+
+ @Override
+ public String getCommand() {
+ return "new-query";
+ }
+
+ @Override
+ public String getDescription() {
+ return "Add a SPARQL query to the Fluo app";
+ }
+
+ @Override
+ public String getUsage() {
+ final JCommander parser = new JCommander(new Parameters());
+
+ final StringBuilder usage = new StringBuilder();
+ parser.usage(usage);
+ return usage.toString();
+ }
+
+ @Override
+ public void execute(final Connector accumulo, final String ryaTablePrefix, final RyaSailRepository rya, final FluoClient fluo, final String[] args) throws ArgumentsException, ExecutionException {
+ checkNotNull(accumulo);
+ checkNotNull(fluo);
+ checkNotNull(args);
+
+ log.trace("Executing the New Query Command...");
+
+ // Parse the command line arguments.
+ final Parameters params = new Parameters();
+ try {
+ new JCommander(params, args);
+ } catch(final ParameterException e) {
+ throw new ArgumentsException("Could not create a new query because of invalid command line parameters.", e);
+ }
+
+ // Load the request from the file into memory.
+ log.trace("Loading the query found in file '" + params.queryRequestFile + "' into the client app.");
+ ParsedQueryRequest request = null;
+ try {
+ final Path requestFile = Paths.get(params.queryRequestFile);
+ final String requestText = IOUtils.toString( Files.newInputStream(requestFile) );
+ request = ParsedQueryRequest.parse(requestText);
+ } catch (final IOException e) {
+ throw new ExecutionException("Could not load the query request into memory.", e);
+ }
+
+ // Load the query into the Fluo app.
+ log.trace("SPARQL Query: " + request.getQuery());
+ log.trace("Var Orders: " + request.getVarOrders());
+ log.trace("Loading these values into the Fluo app.");
+ final CreatePcj createPcj = new CreatePcj();
+ try {
+ createPcj.withRyaIntegration(fluo, ryaTablePrefix, rya, accumulo, request.getVarOrders(), request.getQuery());
+ } catch (MalformedQueryException | SailException | QueryEvaluationException | PcjException e) {
+ throw new ExecutionException("Could not create and load historic matches into the the Fluo app for the query.", e);
+ }
+
+ log.trace("Finished executing the New Query Command.");
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/15ec5d5f/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/command/QueryReportCommand.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/command/QueryReportCommand.java b/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/command/QueryReportCommand.java
new file mode 100644
index 0000000..9a8c7f7
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/command/QueryReportCommand.java
@@ -0,0 +1,104 @@
+/*
+ * 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.client.command;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import org.apache.accumulo.core.client.Connector;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.rya.indexing.pcj.fluo.client.PcjAdminClientCommand;
+import org.apache.rya.indexing.pcj.fluo.api.GetQueryReport;
+import org.apache.rya.indexing.pcj.fluo.api.GetQueryReport.QueryReport;
+import org.apache.rya.indexing.pcj.fluo.client.util.QueryReportRenderer;
+
+import com.beust.jcommander.JCommander;
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.ParameterException;
+
+import io.fluo.api.client.FluoClient;
+import mvm.rya.rdftriplestore.RyaSailRepository;
+
+/**
+ * TODO implement this.
+ */
+public class QueryReportCommand implements PcjAdminClientCommand {
+ private static final Logger log = LogManager.getLogger(NewQueryCommand.class);
+
+ /**
+ * Command line parameters that are used by this command to configure itself.
+ */
+ private static final class Parameters {
+ @Parameter(names = "--queryId", required = true, description = "The Query ID used to build the report.")
+ private String queryId;
+ }
+
+ @Override
+ public String getCommand() {
+ return "query-report";
+ }
+
+ @Override
+ public String getDescription() {
+ return "Build a report that indicates a query's structure within the Fluo app as well as how many Binding Sets have been emitted for each of its nodes.";
+ }
+
+ @Override
+ public String getUsage() {
+ final JCommander parser = new JCommander(new Parameters());
+
+ final StringBuilder usage = new StringBuilder();
+ parser.usage(usage);
+ return usage.toString();
+ }
+
+ @Override
+ public void execute(final Connector accumulo, final String ryaTablePrefix, final RyaSailRepository rya, final FluoClient fluo, final String[] args) throws ArgumentsException, ExecutionException {
+ checkNotNull(accumulo);
+ checkNotNull(ryaTablePrefix);
+ checkNotNull(rya);
+ checkNotNull(fluo);
+ checkNotNull(args);
+
+ log.trace("Executing the Get Query Report Command...");
+
+ // Parse the command line arguments.
+ final Parameters params = new Parameters();
+ try {
+ new JCommander(params, args);
+ } catch(final ParameterException e) {
+ throw new ArgumentsException("Could not create a new query because of invalid command line parameters.", e);
+ }
+
+ // Build the report using what is stored in Fluo.
+ log.trace("Building the report for Query ID: " + params.queryId);
+ final QueryReport queryReport = new GetQueryReport().getReport(fluo, params.queryId);
+ log.trace("Report built.");
+
+ // Format and print the report.
+ try {
+ final String reportString = new QueryReportRenderer().render(queryReport);
+ System.out.println(reportString);
+ } catch(final Exception e) {
+ throw new ExecutionException("Unable to render the query metadata report for output.", e);
+ }
+
+ log.trace("Finished executing the Get Query Report Command.");
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/15ec5d5f/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/FluoLoader.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/FluoLoader.java b/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/FluoLoader.java
new file mode 100644
index 0000000..3168a71
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/FluoLoader.java
@@ -0,0 +1,90 @@
+/*
+ * 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.client.util;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.ArrayList;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.rya.indexing.pcj.fluo.api.InsertTriples;
+import org.openrdf.model.Statement;
+import org.openrdf.rio.RDFHandlerException;
+import org.openrdf.rio.RDFParser;
+import org.openrdf.rio.helpers.RDFHandlerBase;
+
+import io.fluo.api.client.FluoClient;
+import mvm.rya.api.domain.RyaStatement;
+import mvm.rya.api.resolver.RdfToRyaConversions;
+
+/**
+ * When used as the handler of an {@link RDFParser}, instances of this class
+ * will batch load {@link Statement}s into the Fluo app 1000 statements at a time.
+ */
+public class FluoLoader extends RDFHandlerBase {
+ private static final Logger log = LogManager.getLogger(FluoLoader.class);
+
+ private static final int FLUSH_SIZE = 1000;
+ private final ArrayList<RyaStatement> buff = new ArrayList<>(1000);
+
+ private final FluoClient fluoClient;
+ private final InsertTriples insertTriples;
+
+ /**
+ * Constructs an instance of {@link FluoLoader}.
+ *
+ * @param fluoClient - The client that will be used to connect to Fluo. (not null)
+ * @param insertTriples - The interactor that loads triples into a Fluo table. (not null)
+ */
+ public FluoLoader(final FluoClient fluoClient, final InsertTriples insertTriples) {
+ this.fluoClient = checkNotNull(fluoClient);
+ this.insertTriples = checkNotNull(insertTriples);
+ }
+
+ @Override
+ public void startRDF() throws RDFHandlerException {
+ log.trace("Start of RDF file encountered.");
+ }
+
+ @Override
+ public void handleStatement(final Statement st) throws RDFHandlerException {
+ // If the buffer is full, flush it to the Fluo table.
+ if(buff.size() == FLUSH_SIZE) {
+ log.trace("Flushing " + buff.size() + " Statements from the buffer to Fluo.");
+ insertTriples.insert(fluoClient, buff);
+ buff.clear();
+ }
+
+ // Enqueue the statement for the next job.
+ final RyaStatement ryaSt = RdfToRyaConversions.convertStatement(st);
+ buff.add( ryaSt );
+ }
+
+ @Override
+ public void endRDF() throws RDFHandlerException {
+ log.trace("End of RDF file encountered.");
+
+ if(!buff.isEmpty()) {
+ log.trace("Flushing the last " + buff.size() + " Statements from the buffer to Fluo.");
+ insertTriples.insert(fluoClient, buff);
+ buff.clear();
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/15ec5d5f/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/ParsedQueryRequest.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/ParsedQueryRequest.java b/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/ParsedQueryRequest.java
new file mode 100644
index 0000000..252c875
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/ParsedQueryRequest.java
@@ -0,0 +1,134 @@
+/*
+ * 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.client.util;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Objects;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import javax.annotation.ParametersAreNonnullByDefault;
+import javax.annotation.concurrent.Immutable;
+
+import org.apache.commons.lang3.builder.EqualsBuilder;
+
+import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
+
+/**
+ * Represents a request to create a new PCJ in the Fluo app.
+ */
+@Immutable
+@ParametersAreNonnullByDefault
+public class ParsedQueryRequest {
+
+ private final String sparql;
+ private final Set<VariableOrder> varOrders;
+
+ /**
+ * Constructs an instance of {@link CratePcjRequest}.
+ *
+ * @param sparql - The SPARQL query to load into the Fluo app. (not null)
+ * @param varOrders - The variable orders to export the query's results to. (not null)
+ */
+ public ParsedQueryRequest(final String sparql, final Set<VariableOrder> varOrders) {
+ this.sparql = checkNotNull(sparql);
+ this.varOrders = checkNotNull(varOrders);
+ }
+
+ /**
+ * @return The variable orders to export the query's results to. (not null)
+ */
+ public Set<VariableOrder> getVarOrders() {
+ return varOrders;
+ }
+
+ /**
+ * @return The SPARQL query to load into the Fluo app.
+ */
+ public String getQuery() {
+ return sparql;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(sparql, varOrders);
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if(this == o) {
+ return true;
+ }
+ if(o instanceof ParsedQueryRequest) {
+ final ParsedQueryRequest request = (ParsedQueryRequest)o;
+ return new EqualsBuilder()
+ .append(sparql, request.sparql)
+ .append(varOrders, request.varOrders)
+ .isEquals();
+ }
+ return false;
+ }
+
+ private static final Pattern varOrdersPattern = Pattern.compile("^#prefix (.*?)$", Pattern.MULTILINE);
+
+ /**
+ * Create a {@link ParsedQueryRequest} from formatted text.
+ * <p>
+ * The file may contain a list of variable orders at the head as comments.
+ * The file must contain a SPARQL query to loAd into the application.
+ * <p>
+ * Example file:
+ * <pre>
+ * #prefix a, b, c
+ * #prefix b, c, a
+ * SELECT *
+ * WHERE {
+ * ?a <http://talksTo> ?b.
+ * ?b <http://talksTo> ?c.
+ * }
+ * </pre>
+ * @throws IOException
+ */
+ public static ParsedQueryRequest parse(final String requestText) throws IOException {
+ checkNotNull(requestText);
+
+ // Shift this pointer to the end of each match to find where the SPARQL starts.
+ int startOfSparql = 0;
+
+ // Scan for variable orders.
+ final Set<VariableOrder> varOrders = new HashSet<>();
+
+ final Matcher matcher = varOrdersPattern.matcher(requestText);
+ while(matcher.find()) {
+ final String varOrder = matcher.group(1);
+ varOrders.add( new VariableOrder( varOrder.split(",\\s*") ) );
+
+ startOfSparql = matcher.end();
+ }
+
+ // Pull the SPARQL out.
+ final String sparql = requestText.substring(startOfSparql).trim();
+
+ return new ParsedQueryRequest(sparql, varOrders);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/15ec5d5f/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/PcjMetadataRenderer.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/PcjMetadataRenderer.java b/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/PcjMetadataRenderer.java
new file mode 100644
index 0000000..50929bc
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/PcjMetadataRenderer.java
@@ -0,0 +1,100 @@
+/*
+ * 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.client.util;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import java.text.NumberFormat;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rya.indexing.pcj.fluo.client.util.Report.ReportItem;
+import org.openrdf.query.parser.ParsedQuery;
+import org.openrdf.query.parser.sparql.SPARQLParser;
+import org.openrdf.queryrender.sparql.SPARQLQueryRenderer;
+
+import mvm.rya.indexing.external.tupleSet.PcjTables.PcjMetadata;
+import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
+
+/**
+ * Pretty renders the state of a query's {@link PcjMetadata}.
+ */
+public class PcjMetadataRenderer {
+
+ /**
+ * Pretty render the state of a PCJ.
+ *
+ * @param queryId - The ID of the query the metadata is from. (not null)
+ * @param metadata - Metadata about one of the PCJs. (not null)
+ * @return A pretty render of a PCJ's state.
+ * @throws Exception The SPARQL within the metadata could not be parsed or
+ * it could not be rendered.
+ */
+ public String render(final String queryId, final PcjMetadata metadata) throws Exception {
+ checkNotNull(metadata);
+
+ // Pretty format the cardinality.
+ final String cardinality = NumberFormat.getInstance().format( metadata.getCardinality() );
+
+ // Pretty format and split the SPARQL query into lines.
+ final SPARQLParser parser = new SPARQLParser();
+ final SPARQLQueryRenderer renderer = new SPARQLQueryRenderer();
+ final ParsedQuery pq = parser.parseQuery(metadata.getSparql(), null);
+ final String prettySparql = renderer.render(pq);
+ final String[] sparqlLines = StringUtils.split(prettySparql, '\n');
+
+ // Split the variable orders into lines.
+ final String[] varOrderLines = new String[ metadata.getVarOrders().size() ];
+ int i = 0;
+ for(final VariableOrder varOrder : metadata.getVarOrders()) {
+ varOrderLines[i++] = varOrder.toString();
+ }
+
+ // Create the report.
+ final Report.Builder builder = Report.builder();
+ builder.appendItem( new ReportItem("Query ID", queryId) );
+ builder.appendItem( new ReportItem("Cardinality", cardinality) );
+ builder.appendItem( new ReportItem("Export Variable Orders", varOrderLines));
+ builder.appendItem( new ReportItem("SPARQL", sparqlLines) );
+ return builder.build().toString();
+ }
+
+ /**
+ * Pretty render the state of many PCJs.
+ *
+ * @param metadata - The PCJ states to render. The key within the map is
+ * the statem's Query Id. (not null)
+ * @return A pretty render of a PCJs' state.
+ * @throws Exception A SPARQL within the metadata could not be parsed or
+ * it could not be rendered.
+ */
+ public String render(final Map<String, PcjMetadata> metadata) throws Exception {
+ checkNotNull(metadata);
+
+ final StringBuilder formatted = new StringBuilder();
+
+ for(final Entry<String, PcjMetadata> entry : metadata.entrySet()) {
+ final String formattedQuery = render(entry.getKey(), entry.getValue());
+ formatted.append(formattedQuery).append("\n");
+ }
+
+ return formatted.toString();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/15ec5d5f/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/QueryReportRenderer.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/QueryReportRenderer.java b/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/QueryReportRenderer.java
new file mode 100644
index 0000000..2b5bd8a
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/QueryReportRenderer.java
@@ -0,0 +1,112 @@
+/*
+ * 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.client.util;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import javax.annotation.ParametersAreNonnullByDefault;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.rya.indexing.pcj.fluo.api.GetQueryReport.QueryReport;
+import org.apache.rya.indexing.pcj.fluo.app.query.FilterMetadata;
+import org.apache.rya.indexing.pcj.fluo.app.query.FluoQuery;
+import org.apache.rya.indexing.pcj.fluo.app.query.JoinMetadata;
+import org.apache.rya.indexing.pcj.fluo.app.query.QueryMetadata;
+import org.apache.rya.indexing.pcj.fluo.app.query.StatementPatternMetadata;
+import org.apache.rya.indexing.pcj.fluo.client.util.Report.ReportItem;
+import org.openrdf.query.parser.ParsedQuery;
+import org.openrdf.query.parser.sparql.SPARQLParser;
+import org.openrdf.queryrender.sparql.SPARQLQueryRenderer;
+
+/**
+ * Pretty renders a {@link QueryReport}.
+ */
+@ParametersAreNonnullByDefault
+public class QueryReportRenderer {
+
+ /**
+ * Pretty render a {@link QueryReport}.
+ *
+ * @param queryReport - The report that will be rendered. (not null)
+ * @return A pretty render of the report.
+ * @throws Exception Indicates the SPARQL could not be rendered for some reason.
+ */
+ public String render(final QueryReport queryReport) throws Exception {
+ checkNotNull(queryReport);
+
+ final Report.Builder builder = Report.builder();
+
+ final FluoQuery metadata = queryReport.getFluoQuery();
+
+ final QueryMetadata queryMetadata = metadata.getQueryMetadata();
+ builder.appendItem( new ReportItem("QUERY NODE") );
+ builder.appendItem( new ReportItem("Node ID", queryMetadata.getNodeId()) );
+ builder.appendItem( new ReportItem("Variable Order", queryMetadata.getVariableOrder().toString()) );
+ builder.appendItem( new ReportItem("SPARQL", prettyFormatSparql( queryMetadata.getSparql()) ) );
+ builder.appendItem( new ReportItem("Child Node ID", queryMetadata.getChildNodeId()) );
+ builder.appendItem( new ReportItem("Count", "" + queryReport.getCount(queryMetadata.getNodeId())) );
+
+ for(final FilterMetadata filterMetadata : metadata.getFilterMetadata()) {
+ builder.appendItem( new ReportItem("") );
+
+ builder.appendItem( new ReportItem("FILTER NODE") );
+ builder.appendItem( new ReportItem("Node ID", filterMetadata.getNodeId()) );
+ builder.appendItem( new ReportItem("Variable Order", filterMetadata.getVariableOrder().toString()) );
+ builder.appendItem( new ReportItem("Original SPARQL", prettyFormatSparql( filterMetadata.getOriginalSparql()) ) );
+ builder.appendItem( new ReportItem("Filter Index", "" + filterMetadata.getFilterIndexWithinSparql()) );
+ builder.appendItem( new ReportItem("Parent Node ID", filterMetadata.getParentNodeId()) );
+ builder.appendItem( new ReportItem("Child Node ID", filterMetadata.getChildNodeId()) );
+ builder.appendItem( new ReportItem("Count", "" + queryReport.getCount(filterMetadata.getNodeId())) );
+ }
+
+ for(final JoinMetadata joinMetadata : metadata.getJoinMetadata()) {
+ builder.appendItem( new ReportItem("") );
+
+ builder.appendItem( new ReportItem("JOIN NODE") );
+ builder.appendItem( new ReportItem("Node ID", joinMetadata.getNodeId()) );
+ builder.appendItem( new ReportItem("Variable Order", joinMetadata.getVariableOrder().toString()) );
+ builder.appendItem( new ReportItem("Parent Node ID", joinMetadata.getParentNodeId()) );
+ builder.appendItem( new ReportItem("Left Child Node ID", joinMetadata.getLeftChildNodeId()) );
+ builder.appendItem( new ReportItem("Right Child Node ID", joinMetadata.getRightChildNodeId()) );
+ builder.appendItem( new ReportItem("Count", "" + queryReport.getCount(joinMetadata.getNodeId())) );
+ }
+
+ for(final StatementPatternMetadata spMetadata : metadata.getStatementPatternMetadata()) {
+ builder.appendItem( new ReportItem("") );
+
+ builder.appendItem( new ReportItem("STATEMENT PATTERN NODE") );
+ builder.appendItem( new ReportItem("Node ID", spMetadata.getNodeId()) );
+ builder.appendItem( new ReportItem("Variable Order", spMetadata.getVariableOrder().toString()) );
+ builder.appendItem( new ReportItem("Statement Pattern", spMetadata.getStatementPattern()) );
+ builder.appendItem( new ReportItem("Parent Node ID", spMetadata.getParentNodeId()) );
+ builder.appendItem( new ReportItem("Count", "" + queryReport.getCount(spMetadata.getNodeId())) );
+ }
+
+ return builder.build().toString();
+ }
+
+ private String[] prettyFormatSparql(final String sparql) throws Exception {
+ final SPARQLParser parser = new SPARQLParser();
+ final SPARQLQueryRenderer renderer = new SPARQLQueryRenderer();
+ final ParsedQuery pq = parser.parseQuery(sparql, null);
+ final String prettySparql = renderer.render(pq);
+ final String[] sparqlLines = StringUtils.split(prettySparql, '\n');
+ return sparqlLines;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/15ec5d5f/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/Report.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/Report.java b/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/Report.java
new file mode 100644
index 0000000..cdd51f8
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/Report.java
@@ -0,0 +1,214 @@
+/*
+ * 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.client.util;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import javax.annotation.ParametersAreNonnullByDefault;
+import javax.annotation.concurrent.Immutable;
+
+import org.apache.commons.lang3.StringUtils;
+
+import com.google.common.collect.ImmutableList;
+
+/**
+ * A human readable report that displays the title of a {@link ReportItem} on the
+ * left hand side of the table and the value of the item on the right hand side.
+ * If an item does not have any values, then it prints an empty line.
+ */
+@Immutable
+@ParametersAreNonnullByDefault
+public class Report {
+
+ private final ImmutableList<ReportItem> items;
+ private final int maxTitleLength;
+ private final int maxValueLineLength;
+
+ /**
+ * Use an instance of {@link Report.Builder} to construct instances of this class.
+ *
+ * @param items - An ordered list of items that appear in the report. (not null)
+ * @param maxTitleLength - The length of the longest title in the report. (> 0)
+ * @param maxValueLineLength - The length of the longest value line in the report. (> 0)
+ */
+ private Report(
+ final ImmutableList<ReportItem> items,
+ final int maxTitleLength,
+ final int maxValueLineLength) {
+ this.items = checkNotNull(items);
+ checkArgument(maxTitleLength > 0);
+ this.maxTitleLength = maxTitleLength;
+ checkArgument(maxValueLineLength > 0);
+ this.maxValueLineLength = maxValueLineLength;
+ }
+
+ @Override
+ public String toString() {
+ // Figure out how long each line will be.
+ final int lineLength = "| ".length() + maxTitleLength + " | ".length() + maxValueLineLength + " |".length();
+
+ // Format that may be used to write each line.
+ final String lineFormat = "| %-" + maxTitleLength + "s | %-" + maxValueLineLength + "s |\n";
+
+ // The line that is used as the first and last line of the report.
+ final String dashLine = StringUtils.repeat("-", lineLength);
+
+ // Build the String verison of the report.
+ final StringBuilder builder = new StringBuilder();
+ builder.append(dashLine).append("\n");
+
+ for(final ReportItem item : items) {
+ final String[] valueLines = item.getValueLines();
+ switch(valueLines.length) {
+ case 0:
+ // Write an empty value cell.
+ builder.append( String.format(lineFormat, item.getTitle(), "") );
+ break;
+
+ case 1:
+ // Write the value cell.
+ builder.append( String.format(lineFormat, item.getTitle(), valueLines[0]) );
+ break;
+
+ default:
+ builder.append( String.format(lineFormat, item.getTitle(), valueLines[0]) );
+ for(int i = 1; i < valueLines.length; i++) {
+ builder.append( String.format(lineFormat, "", valueLines[i]) );
+ }
+ break;
+ }
+ }
+
+ builder.append(dashLine).append("\n");
+ return builder.toString();
+ }
+
+ /**
+ * @return Creates a new {@link Report.Builder}.
+ */
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ /**
+ * An item that may appear within a {@link Report}. Each item has a title
+ * that briefly describes the value it holds.
+ */
+ @Immutable
+ @ParametersAreNonnullByDefault
+ public static final class ReportItem {
+ private final String title;
+ private final String[] valueLines;
+
+ /**
+ * Constructs an instance of {@link ReportItem} when there is no value
+ * associated with the title.
+ *
+ * @param title - Describes the item's value. (not null)
+ */
+ public ReportItem(final String title) {
+ this.title = checkNotNull(title);
+ this.valueLines = new String[0];
+ }
+
+ /**
+ * Constructs an instance of {@link ReportItem} when the value section
+ * is contained within a single line.
+ *
+ * @param title - Describes the item's value. (not null)
+ * @param valueLine - The line that will appear within the value section. (not null)
+ */
+ public ReportItem(final String title, final String valueLine) {
+ this.title = checkNotNull(title);
+ checkNotNull(valueLine);
+ this.valueLines = new String[]{ valueLine };
+ }
+
+ /**
+ * Constructs an instance of {@link ReportItem} when the value section
+ * spans many lines.
+ *
+ * @param title - Describes the item's value. (not null)
+ * @param valueLines - The value section broken into lines as they
+ * will appear within the report. (not null)
+ */
+ public ReportItem(final String title, final String[] valueLines) {
+ this.title = checkNotNull(title);
+ this.valueLines = checkNotNull(valueLines);
+ }
+
+ /**
+ * @return Describes the item's value.
+ */
+ public String getTitle() {
+ return title;
+ }
+
+ /**
+ * @return The value section broken into lines as they will appear within the report.
+ */
+ public String[] getValueLines() {
+ return valueLines;
+ }
+ }
+
+ /**
+ * Builds instances of {@link Report}.
+ */
+ @ParametersAreNonnullByDefault
+ public static final class Builder {
+
+ private final ImmutableList.Builder<ReportItem> lines = ImmutableList.builder();
+ private int maxTitleLength = 0;
+ private int maxValueLineLength = 0;
+
+ /**
+ * Append a {@link ReportItem} to the end of the {@link Report}.
+ *
+ * @param item - The next item that will appear in the report. (not null)
+ * @return This builder so that method invocations may be chained.
+ */
+ public Builder appendItem(final ReportItem item) {
+ checkNotNull(item);
+ lines.add(item);
+
+ final int titleLength = item.getTitle().length();
+ if(maxTitleLength < titleLength) {
+ this.maxTitleLength = titleLength;
+ }
+
+ for(final String valueLine : item.getValueLines()) {
+ final int valueLineLength = valueLine.length();
+ if(maxValueLineLength < valueLineLength) {
+ this.maxValueLineLength = valueLineLength;
+ }
+ }
+
+ return this;
+ }
+
+ /**
+ * @return An instance of {@link Report} using the state of the builder.
+ */
+ public Report build() {
+ return new Report( lines.build(), maxTitleLength, maxValueLineLength);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/15ec5d5f/extras/rya.pcj.fluo/pcj.fluo.client/src/test/java/org/apache/rya/indexing/pcj/fluo/client/ParsedQueryRequestTest.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.client/src/test/java/org/apache/rya/indexing/pcj/fluo/client/ParsedQueryRequestTest.java b/extras/rya.pcj.fluo/pcj.fluo.client/src/test/java/org/apache/rya/indexing/pcj/fluo/client/ParsedQueryRequestTest.java
new file mode 100644
index 0000000..432cf8a
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.client/src/test/java/org/apache/rya/indexing/pcj/fluo/client/ParsedQueryRequestTest.java
@@ -0,0 +1,83 @@
+/*
+ * 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.client;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import java.util.HashSet;
+
+import org.apache.rya.indexing.pcj.fluo.client.util.ParsedQueryRequest;
+import org.junit.Test;
+
+import com.google.common.collect.Sets;
+
+import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
+
+/**
+ * Tests the methods of {@link ParsedQueryRequest}.
+ */
+public class ParsedQueryRequestTest {
+
+ @Test
+ public void parseNoVarOrders() throws IOException {
+ final String requestText =
+ "SELECT * \n"+
+ "WHERE { \n" +
+ " ?a <http://talksTo> ?b. \n" +
+ " ?b <http://talksTo> ?c. \n" +
+ "}";
+
+ final ParsedQueryRequest expected = new ParsedQueryRequest(
+ "SELECT * \n"+
+ "WHERE { \n" +
+ " ?a <http://talksTo> ?b. \n" +
+ " ?b <http://talksTo> ?c. \n" +
+ "}",
+ new HashSet<VariableOrder>());
+
+ final ParsedQueryRequest request = ParsedQueryRequest.parse(requestText);
+ assertEquals(expected, request);
+ }
+
+ @Test
+ public void parseHasVarOrders() throws IOException {
+ final String requestText =
+ "#prefix a, b,c\n" +
+ "#prefix b, c, a\n" +
+ "SELECT * \n"+
+ "WHERE { \n" +
+ " ?a <http://talksTo> ?b. \n" +
+ " ?b <http://talksTo> ?c. \n" +
+ "}";
+
+ final ParsedQueryRequest expected = new ParsedQueryRequest(
+ "SELECT * \n"+
+ "WHERE { \n" +
+ " ?a <http://talksTo> ?b. \n" +
+ " ?b <http://talksTo> ?c. \n" +
+ "}",
+ Sets.newHashSet(
+ new VariableOrder("a", "b", "c"),
+ new VariableOrder("b", "c", "a")));
+
+ final ParsedQueryRequest request = ParsedQueryRequest.parse(requestText);
+ assertEquals(expected, request);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/15ec5d5f/extras/rya.pcj.fluo/pcj.fluo.client/src/test/java/org/apache/rya/indexing/pcj/fluo/client/PcjMetadataRendererTest.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.client/src/test/java/org/apache/rya/indexing/pcj/fluo/client/PcjMetadataRendererTest.java b/extras/rya.pcj.fluo/pcj.fluo.client/src/test/java/org/apache/rya/indexing/pcj/fluo/client/PcjMetadataRendererTest.java
new file mode 100644
index 0000000..a200039
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.client/src/test/java/org/apache/rya/indexing/pcj/fluo/client/PcjMetadataRendererTest.java
@@ -0,0 +1,125 @@
+/*
+ * 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.client;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.apache.rya.indexing.pcj.fluo.client.util.PcjMetadataRenderer;
+import org.junit.Test;
+
+import com.google.common.collect.Sets;
+
+import mvm.rya.indexing.external.tupleSet.PcjTables.PcjMetadata;
+import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
+
+/**
+ * Tests the methods of {@link PcjMetadataRenderer}.
+ */
+public class PcjMetadataRendererTest {
+
+ @Test
+ public void formatSingleMetadata() throws Exception {
+ // Create the PcjMetadata that will be formatted as a report.
+ final PcjMetadata metadata = new PcjMetadata(
+ "SELECT ?x ?y " +
+ "WHERE { " +
+ "?x <http://talksTo> <http://Eve>. " +
+ "?y <http://worksAt> <http://Chipotle>." +
+ "}",
+ 12233423L,
+ Sets.<VariableOrder>newHashSet(
+ new VariableOrder("x", "y"),
+ new VariableOrder("y", "x")));
+
+ // Run the test.
+ final String expected =
+ "---------------------------------------------------------------------\n" +
+ "| Query ID | query1 |\n" +
+ "| Cardinality | 12,233,423 |\n" +
+ "| Export Variable Orders | y;x |\n" +
+ "| | x;y |\n" +
+ "| SPARQL | select ?x ?y |\n" +
+ "| | where { |\n" +
+ "| | ?x <http://talksTo> <http://Eve>. |\n" +
+ "| | ?y <http://worksAt> <http://Chipotle>. |\n" +
+ "| | } |\n" +
+ "---------------------------------------------------------------------\n";
+
+ final PcjMetadataRenderer formatter = new PcjMetadataRenderer();
+ assertEquals(expected, formatter.render("query1", metadata));
+ }
+
+ @Test
+ public void formatManyMetdata() throws Exception {
+ // Create the PcjMetadata that will be formatted as a report.
+ final PcjMetadata metadata1 = new PcjMetadata(
+ "SELECT ?x ?y " +
+ "WHERE { " +
+ "?x <http://talksTo> <http://Eve>. " +
+ "?y <http://worksAt> <http://Chipotle>." +
+ "}",
+ 12233423L,
+ Sets.<VariableOrder>newHashSet(
+ new VariableOrder("x", "y"),
+ new VariableOrder("y", "x")));
+
+ final PcjMetadata metadata2 = new PcjMetadata(
+ "SELECT ?x " +
+ "WHERE { " +
+ "?x <http://likes> <http://cookies>" +
+ "}",
+ 2342L,
+ Sets.<VariableOrder>newHashSet(new VariableOrder("x")));
+
+ final Map<String, PcjMetadata> metadata = new LinkedHashMap<>();
+ metadata.put("query1", metadata1);
+ metadata.put("query2", metadata2);
+
+ // Run the test.
+ final String expected =
+ "---------------------------------------------------------------------\n" +
+ "| Query ID | query1 |\n" +
+ "| Cardinality | 12,233,423 |\n" +
+ "| Export Variable Orders | y;x |\n" +
+ "| | x;y |\n" +
+ "| SPARQL | select ?x ?y |\n" +
+ "| | where { |\n" +
+ "| | ?x <http://talksTo> <http://Eve>. |\n" +
+ "| | ?y <http://worksAt> <http://Chipotle>. |\n" +
+ "| | } |\n" +
+ "---------------------------------------------------------------------\n" +
+ "\n" +
+ "------------------------------------------------------------------\n" +
+ "| Query ID | query2 |\n" +
+ "| Cardinality | 2,342 |\n" +
+ "| Export Variable Orders | x |\n" +
+ "| SPARQL | select ?x |\n" +
+ "| | where { |\n" +
+ "| | ?x <http://likes> <http://cookies>. |\n" +
+ "| | } |\n" +
+ "------------------------------------------------------------------\n" +
+ "\n";
+
+ final PcjMetadataRenderer formatter = new PcjMetadataRenderer();
+ assertEquals(expected, formatter.render(metadata));
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/15ec5d5f/extras/rya.pcj.fluo/pcj.fluo.client/src/test/java/org/apache/rya/indexing/pcj/fluo/client/ReportTests.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.client/src/test/java/org/apache/rya/indexing/pcj/fluo/client/ReportTests.java b/extras/rya.pcj.fluo/pcj.fluo.client/src/test/java/org/apache/rya/indexing/pcj/fluo/client/ReportTests.java
new file mode 100644
index 0000000..25e21e0
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.client/src/test/java/org/apache/rya/indexing/pcj/fluo/client/ReportTests.java
@@ -0,0 +1,82 @@
+/*
+ * 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.client;
+
+import static org.junit.Assert.assertEquals;
+
+import org.apache.rya.indexing.pcj.fluo.client.util.Report;
+import org.apache.rya.indexing.pcj.fluo.client.util.Report.ReportItem;
+import org.junit.Test;
+
+/**
+ * Tests the methods of {@link Report}.
+ */
+public class ReportTests {
+
+ @Test
+ public void singleLineValues() {
+ final Report.Builder builder = Report.builder();
+ builder.appendItem(new ReportItem("Title 1", new String[]{"Short value."}));
+ builder.appendItem(new ReportItem("Title 2", new String[]{"This is the longest values that appears in the report."}));
+ builder.appendItem(new ReportItem("This is a long title", new String[]{"Short value."}));
+ final Report report = builder.build();
+
+ final String expected =
+ "---------------------------------------------------------------------------------\n" +
+ "| Title 1 | Short value. |\n" +
+ "| Title 2 | This is the longest values that appears in the report. |\n" +
+ "| This is a long title | Short value. |\n" +
+ "---------------------------------------------------------------------------------\n";
+
+ assertEquals(expected, report.toString());
+ }
+
+ @Test
+ public void emptyValues() {
+ final Report.Builder builder = Report.builder();
+ builder.appendItem(new ReportItem("No Value Here", new String[]{}));
+ builder.appendItem(new ReportItem("Value Here", new String[]{"This one has a value."}));
+ final Report report = builder.build();
+
+ final String expected =
+ "-----------------------------------------\n" +
+ "| No Value Here | |\n" +
+ "| Value Here | This one has a value. |\n" +
+ "-----------------------------------------\n";
+ assertEquals(expected, report.toString());
+ }
+
+ @Test
+ public void multiLineValues() {
+ final Report.Builder builder = Report.builder();
+ builder.appendItem(new ReportItem("Title 1", new String[]{"Value 1"}));
+ builder.appendItem(new ReportItem("Multiple Lines", new String[]{"This is the first line.", "This is the second line."}));
+ builder.appendItem(new ReportItem("Title 2", new String[]{"Value 2"}));
+ final Report report = builder.build();
+
+ final String expected =
+ "---------------------------------------------\n" +
+ "| Title 1 | Value 1 |\n" +
+ "| Multiple Lines | This is the first line. |\n" +
+ "| | This is the second line. |\n" +
+ "| Title 2 | Value 2 |\n" +
+ "---------------------------------------------\n";
+ assertEquals(expected, report.toString());
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/15ec5d5f/extras/rya.pcj.fluo/pcj.fluo.demo/pom.xml
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.demo/pom.xml b/extras/rya.pcj.fluo/pcj.fluo.demo/pom.xml
new file mode 100644
index 0000000..97a967b
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.demo/pom.xml
@@ -0,0 +1,85 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <parent>
+ <groupId>org.apache.rya</groupId>
+ <artifactId>rya.pcj.fluo.parent</artifactId>
+ <version>3.2.10-SNAPSHOT</version>
+ </parent>
+
+ <modelVersion>4.0.0</modelVersion>
+ <artifactId>rya.pcj.fluo.demo</artifactId>
+
+ <name>Apache Rya PCJ Fluo Demo</name>
+ <description>
+ A demo application that shows how the Fluo application is able to maintain
+ Rya Precomputed Joins using historic Statements while streaming new Statements.
+ </description>
+
+ <dependencies>
+ <!-- Rya Runtime Dependencies. -->
+ <dependency>
+ <groupId>org.apache.rya</groupId>
+ <artifactId>rya.pcj.fluo.api</artifactId>
+ </dependency>
+
+ <!-- 3rd Party Runtime Dependencies. -->
+ <dependency>
+ <groupId>org.openrdf.sesame</groupId>
+ <artifactId>sesame-queryrender</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>io.fluo</groupId>
+ <artifactId>fluo-mini</artifactId>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <!-- Use the pre-build 'jar-with-dependencies' assembly to package the dependent class files into the final jar.
+ This creates a jar file that can be deployed to Fluo without having to include any dependent jars. -->
+ <plugin>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <configuration>
+ <archive>
+ <manifest>
+ <mainClass>org.apache.rya.indexing.pcj.fluo.demo.DemoDriver</mainClass>
+ </manifest>
+ </archive>
+ <descriptorRefs>
+ <descriptorRef>jar-with-dependencies</descriptorRef>
+ </descriptorRefs>
+ </configuration>
+ <executions>
+ <execution>
+ <id>make-assembly</id>
+ <phase>package</phase>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/15ec5d5f/extras/rya.pcj.fluo/pcj.fluo.demo/src/main/java/org/apache/rya/indexing/pcj/fluo/demo/Demo.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.demo/src/main/java/org/apache/rya/indexing/pcj/fluo/demo/Demo.java b/extras/rya.pcj.fluo/pcj.fluo.demo/src/main/java/org/apache/rya/indexing/pcj/fluo/demo/Demo.java
new file mode 100644
index 0000000..f1ec3dd
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.demo/src/main/java/org/apache/rya/indexing/pcj/fluo/demo/Demo.java
@@ -0,0 +1,63 @@
+/*
+ * 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.demo;
+
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.minicluster.MiniAccumuloCluster;
+import org.openrdf.repository.RepositoryConnection;
+
+import io.fluo.api.client.FluoClient;
+import io.fluo.api.mini.MiniFluo;
+import mvm.rya.rdftriplestore.RyaSailRepository;
+
+/**
+ * Represents a demonstration that uses Rya and Fluo on top of Accumulo.
+ */
+public interface Demo {
+
+ /**
+ * Run the demo.
+ */
+ public void execute(
+ MiniAccumuloCluster accumulo,
+ Connector accumuloConn,
+ String ryaTablePrefix,
+ RyaSailRepository ryaRepo,
+ RepositoryConnection ryaConn,
+ MiniFluo fluo,
+ FluoClient fluoClient) throws DemoExecutionException;
+
+ /**
+ * A {@link Demo}'s execution could not be completed because of a non-recoverable
+ * problem while running.
+ */
+ public static class DemoExecutionException extends Exception {
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * Constructs an instance of {@link }.
+ *
+ * @param message - Describes why this is being thrown.
+ * @param cause - The exception that caused this one to be thrown.
+ */
+ public DemoExecutionException(final String message, final Exception cause) {
+ super(message, cause);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/15ec5d5f/extras/rya.pcj.fluo/pcj.fluo.demo/src/main/java/org/apache/rya/indexing/pcj/fluo/demo/DemoDriver.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.demo/src/main/java/org/apache/rya/indexing/pcj/fluo/demo/DemoDriver.java b/extras/rya.pcj.fluo/pcj.fluo.demo/src/main/java/org/apache/rya/indexing/pcj/fluo/demo/DemoDriver.java
new file mode 100644
index 0000000..d985b06
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.demo/src/main/java/org/apache/rya/indexing/pcj/fluo/demo/DemoDriver.java
@@ -0,0 +1,289 @@
+/*
+ * 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.demo;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Instance;
+import org.apache.accumulo.core.client.ZooKeeperInstance;
+import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.minicluster.MiniAccumuloCluster;
+import org.apache.log4j.ConsoleAppender;
+import org.apache.log4j.Level;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+import org.apache.log4j.PatternLayout;
+import org.apache.rya.indexing.pcj.fluo.app.export.rya.RyaExportParameters;
+import org.apache.rya.indexing.pcj.fluo.app.observers.FilterObserver;
+import org.apache.rya.indexing.pcj.fluo.app.observers.JoinObserver;
+import org.apache.rya.indexing.pcj.fluo.app.observers.QueryResultObserver;
+import org.apache.rya.indexing.pcj.fluo.app.observers.StatementPatternObserver;
+import org.apache.rya.indexing.pcj.fluo.app.observers.TripleObserver;
+import org.apache.rya.indexing.pcj.fluo.demo.Demo.DemoExecutionException;
+import org.openrdf.repository.RepositoryConnection;
+import org.openrdf.repository.RepositoryException;
+
+import com.google.common.io.Files;
+
+import io.fluo.api.client.FluoClient;
+import io.fluo.api.client.FluoFactory;
+import io.fluo.api.config.FluoConfiguration;
+import io.fluo.api.config.ObserverConfiguration;
+import io.fluo.api.mini.MiniFluo;
+import mvm.rya.accumulo.AccumuloRdfConfiguration;
+import mvm.rya.accumulo.AccumuloRyaDAO;
+import mvm.rya.api.RdfCloudTripleStoreConfiguration;
+import mvm.rya.indexing.accumulo.ConfigUtils;
+import mvm.rya.rdftriplestore.RdfCloudTripleStore;
+import mvm.rya.rdftriplestore.RyaSailRepository;
+
+/**
+ * Runs {@link Demo}s that require Rya and Fluo.
+ */
+public class DemoDriver {
+ private static final Logger log = Logger.getLogger(DemoDriver.class);
+
+ private static final String RYA_TABLE_PREFIX = "demo_";
+
+ // Rya data store and connections.
+ private static MiniAccumuloCluster accumulo = null;
+ private static Connector accumuloConn = null;
+ private static RyaSailRepository ryaRepo = null;
+ private static RepositoryConnection ryaConn = null;
+
+ // Fluo data store and connections.
+ private static MiniFluo fluo = null;
+ private static FluoClient fluoClient = null;
+
+ public static void main(final String[] args) {
+ setupLogging();
+
+ // Setup the resources required to run the demo.
+ try {
+ log.info("Initializing resources used by the demo...");
+ setupResources();
+ } catch (final DemoInitializationException e) {
+ log.error("Could not initialize the demo's resources. Exiting.", e);
+ System.exit(-1);
+ }
+ log.info("");
+
+ // Run the demo.
+ try {
+ new FluoAndHistoricPcjsDemo().execute(accumulo, accumuloConn, RYA_TABLE_PREFIX, ryaRepo, ryaConn, fluo, fluoClient);
+ } catch (final DemoExecutionException e) {
+ log.error("An exception was thrown durring demo execution. The demo can not continue.", e);
+ }
+
+ // Tear down the demo environment.
+ log.info("Shutting down the demo...");
+ shutdownResources();
+ log.info("Demo exiting.");
+ }
+
+ private static void setupLogging() {
+ // Turn off all the loggers and customize how they write to the console.
+ final Logger rootLogger = LogManager.getRootLogger();
+ rootLogger.setLevel(Level.OFF);
+ final ConsoleAppender ca = (ConsoleAppender) rootLogger.getAppender("stdout");
+ ca.setLayout(new PatternLayout("%-5p - %m%n"));
+
+
+ // Turn the loggers used by the demo back on.
+ log.setLevel(Level.INFO);
+ }
+
+ /**
+ * Indicates a problem while initializing the demo's resources prevented it from starting.
+ */
+ private static final class DemoInitializationException extends Exception {
+ private static final long serialVersionUID = 1L;
+
+ public DemoInitializationException(final String message, final Exception cause) {
+ super(message, cause);
+ }
+ }
+
+ private static void setupResources() throws DemoInitializationException {
+ try{
+ // Initialize the Mini Accumulo that will be used to store Triples and get a connection to it.
+ log.debug("Starting up the Mini Accumulo Cluster used by Rya.");
+ accumulo = startMiniAccumulo();
+
+ // Setup the Rya library to use the Mini Accumulo.
+ log.debug("Starting up the Rya Repository.");
+ ryaRepo = setupRya(accumulo);
+ ryaConn = ryaRepo.getConnection();
+
+ // Initialize the Mini Fluo that will be used to store created queries.
+ log.debug("Starting up the Mini Fluo instance.");
+ fluo = startMiniFluo();
+ fluoClient = FluoFactory.newClient( fluo.getClientConfiguration() );
+ } catch(final Exception e) {
+ throw new DemoInitializationException("Could not run the demo because of a problem while initializing the mini resources.", e);
+ }
+ }
+
+ private static void shutdownResources() {
+ if(ryaConn != null) {
+ try {
+ log.debug("Shutting down Rya Connection.");
+ ryaConn.close();
+ } catch(final Exception e) {
+ log.error("Could not shut down the Rya Connection.", e);
+ }
+ }
+
+ if(ryaRepo != null) {
+ try {
+ log.debug("Shutting down Rya Repo.");
+ ryaRepo.shutDown();
+ } catch(final Exception e) {
+ log.error("Could not shut down the Rya Repo.", e);
+ }
+ }
+
+ if(accumulo != null) {
+ try {
+ log.debug("Shutting down the Mini Accumulo being used as a Rya store.");
+ accumulo.stop();
+ } catch(final Exception e) {
+ log.error("Could not shut down the Mini Accumulo.", e);
+ }
+ }
+
+ if(fluoClient != null) {
+ try {
+ log.debug("Shutting down Fluo Client.");
+ fluoClient.close();
+ } catch(final Exception e) {
+ log.error("Could not shut down the Fluo Client.", e);
+ }
+ }
+
+ if(fluo != null) {
+ try {
+ log.debug("Shutting down Mini Fluo.");
+ fluo.close();
+ } catch (final Exception e) {
+ log.error("Could not shut down the Mini Fluo.", e);
+ }
+ }
+ }
+
+ /**
+ * Setup a Mini Accumulo cluster that uses a temporary directory to store its data.
+ *
+ * @return A Mini Accumulo cluster.
+ */
+ private static MiniAccumuloCluster startMiniAccumulo() throws IOException, InterruptedException, AccumuloException, AccumuloSecurityException {
+ final File miniDataDir = Files.createTempDir();
+
+ // Setup and start the Mini Accumulo.
+ final MiniAccumuloCluster accumulo = new MiniAccumuloCluster(miniDataDir, "password");
+ accumulo.start();
+
+ // Store a connector to the Mini Accumulo.
+ final Instance instance = new ZooKeeperInstance(accumulo.getInstanceName(), accumulo.getZooKeepers());
+ accumuloConn = instance.getConnector("root", new PasswordToken("password"));
+
+ return accumulo;
+ }
+
+ /**
+ * Format a Mini Accumulo to be a Rya repository.
+ *
+ * @param accumulo - The Mini Accumulo cluster Rya will sit on top of. (not null)
+ * @return The Rya repository sitting on top of the Mini Accumulo.
+ */
+ private static RyaSailRepository setupRya(final MiniAccumuloCluster accumulo) throws AccumuloException, AccumuloSecurityException, RepositoryException {
+ checkNotNull(accumulo);
+
+ // Setup the Rya Repository that will be used to create Repository Connections.
+ final RdfCloudTripleStore ryaStore = new RdfCloudTripleStore();
+ final AccumuloRyaDAO crdfdao = new AccumuloRyaDAO();
+ crdfdao.setConnector(accumuloConn);
+
+ // Setup Rya configuration values.
+ final AccumuloRdfConfiguration conf = new AccumuloRdfConfiguration();
+ conf.setTablePrefix("demo_");
+ conf.setDisplayQueryPlan(true);
+
+ conf.setBoolean(ConfigUtils.USE_MOCK_INSTANCE, true);
+ conf.set(RdfCloudTripleStoreConfiguration.CONF_TBL_PREFIX, RYA_TABLE_PREFIX);
+ conf.set(ConfigUtils.CLOUDBASE_USER, "root");
+ conf.set(ConfigUtils.CLOUDBASE_PASSWORD, "password");
+ conf.set(ConfigUtils.CLOUDBASE_INSTANCE, accumulo.getInstanceName());
+
+ crdfdao.setConf(conf);
+ ryaStore.setRyaDAO(crdfdao);
+
+ final RyaSailRepository ryaRepo = new RyaSailRepository(ryaStore);
+ ryaRepo.initialize();
+
+ return ryaRepo;
+ }
+
+ /**
+ * Setup a Mini Fluo cluster that uses a temporary directory to store its data.ll
+ *
+ * @return A Mini Fluo cluster.
+ */
+ private static MiniFluo startMiniFluo() {
+ final File miniDataDir = Files.createTempDir();
+
+ // Setup the observers that will be used by the Fluo PCJ Application.
+ final List<ObserverConfiguration> observers = new ArrayList<>();
+ observers.add(new ObserverConfiguration(TripleObserver.class.getName()));
+ observers.add(new ObserverConfiguration(StatementPatternObserver.class.getName()));
+ observers.add(new ObserverConfiguration(JoinObserver.class.getName()));
+ observers.add(new ObserverConfiguration(FilterObserver.class.getName()));
+
+ // Provide export parameters child test classes may provide to the export observer.
+ final HashMap<String, String> exportParams = new HashMap<>();
+ final RyaExportParameters ryaParams = new RyaExportParameters(exportParams);
+ ryaParams.setExportToRya(true);
+ ryaParams.setAccumuloInstanceName(accumulo.getInstanceName());
+ ryaParams.setZookeeperServers(accumulo.getZooKeepers());
+ ryaParams.setExporterUsername("root");
+ ryaParams.setExporterPassword("password");
+
+ final ObserverConfiguration exportObserverConfig = new ObserverConfiguration(QueryResultObserver.class.getName());
+ exportObserverConfig.setParameters( exportParams );
+ observers.add(exportObserverConfig);
+
+ // Configure how the mini fluo will run.
+ final FluoConfiguration config = new FluoConfiguration();
+ config.setApplicationName("IntegrationTests");
+ config.setMiniDataDir(miniDataDir.getAbsolutePath());
+ config.addObservers(observers);
+
+ final MiniFluo miniFluo = FluoFactory.newMiniFluo(config);
+ return miniFluo;
+ }
+}
\ No newline at end of file