extras/rya.pcj.fluo/pcj.fluo.api/src/main/java/org/apache/rya/indexing/pcj/fluo/api/CreateFluoPcj.java - rya - Git at Google

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

 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static java.util.Objects.requireNonNull;

 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;

 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.security.Authorizations;
 import org.apache.fluo.api.client.FluoClient;
 import org.apache.fluo.api.client.Transaction;
 import org.apache.log4j.Logger;
 import org.apache.rya.accumulo.AccumuloRdfConfiguration;
 import org.apache.rya.accumulo.query.AccumuloRyaQueryEngine;
 import org.apache.rya.api.client.CreatePCJ.ExportStrategy;
 import org.apache.rya.api.client.CreatePCJ.QueryType;
 import org.apache.rya.api.domain.RyaStatement;
 import org.apache.rya.api.domain.RyaType;
 import org.apache.rya.api.domain.RyaURI;
 import org.apache.rya.api.persist.RyaDAOException;
 import org.apache.rya.api.persist.query.BatchRyaQuery;
 import org.apache.rya.api.resolver.RdfToRyaConversions;
 import org.apache.rya.indexing.pcj.fluo.app.FluoStringConverter;
 import org.apache.rya.indexing.pcj.fluo.app.NodeType;
 import org.apache.rya.indexing.pcj.fluo.app.query.FluoQuery;
 import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryMetadataDAO;
 import org.apache.rya.indexing.pcj.fluo.app.query.SparqlFluoQueryBuilder;
 import org.apache.rya.indexing.pcj.fluo.app.query.StatementPatternMetadata;
 import org.apache.rya.indexing.pcj.fluo.app.query.UnsupportedQueryException;
 import org.apache.rya.indexing.pcj.fluo.app.util.FluoQueryUtils;
 import org.apache.rya.indexing.pcj.storage.PcjException;
 import org.apache.rya.indexing.pcj.storage.PcjMetadata;
 import org.apache.rya.indexing.pcj.storage.PrecomputedJoinStorage;
 import org.calrissian.mango.collect.CloseableIterable;
 import org.eclipse.rdf4j.model.IRI;
 import org.eclipse.rdf4j.model.Resource;
 import org.eclipse.rdf4j.model.Value;
 import org.eclipse.rdf4j.query.MalformedQueryException;
 import org.eclipse.rdf4j.query.algebra.StatementPattern;

 import com.google.common.base.Preconditions;
 import com.google.common.collect.Sets;

 import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
 import edu.umd.cs.findbugs.annotations.NonNull;

 /**
  * Sets up a new Pre Computed Join (PCJ) in Fluo from a SPARQL query.
  * <p>
  * This is a two phase process.
  * <ol>
  *   <li>Setup metadata about each node of the query using a single Fluo transaction. </li>
  *   <li>Scan Rya for binding sets that match each Statement Pattern from the query
  *       and use a separate Fluo transaction for each batch that is inserted. This
  *       ensure historic triples will be included in the query's results.</li>
  * </ol>
  * After the first step is finished, any new Triples that are added to the Fluo
  * application will be matched against statement patterns, the final results
  * will percolate to the top of the query, and those results will be exported to
  * Rya's query system.
  */
 @DefaultAnnotation(NonNull.class)
 public class CreateFluoPcj {
     private static final Logger log = Logger.getLogger(CreateFluoPcj.class);

     /**
      * The default Statement Pattern batch insert size is 1000.
      */
     private static final int DEFAULT_SP_INSERT_BATCH_SIZE = 1000;
     /**
      * The default Join batch size is 5000.
      */
     private static final int DEFAULT_JOIN_BATCH_SIZE = 5000;

     /**
      * The maximum number of binding sets that will be inserted into each Statement
      * Pattern's result set per Fluo transaction.
      */
     private final int spInsertBatchSize;

     /**
      * The maximum number of join results that will be processed per transaction.
      */
     private final int joinBatchSize;
     /**
      * Constructs an instance of {@link CreateFluoPcj} that uses
      * {@link #DEFAULT_SP_INSERT_BATCH_SIZE} as the default batch insert size.
      */
     public CreateFluoPcj() {
         this(DEFAULT_SP_INSERT_BATCH_SIZE, DEFAULT_JOIN_BATCH_SIZE);
     }

     /**
      * Constructs an instance of {@link CreateFluoPcj}.
      *
      * @param spInsertBatchSize - The maximum number of binding sets that will be
      *   inserted into each Statement Pattern's result set per Fluo transaction.
      */
     public CreateFluoPcj(final int spInsertBatchSize, final int joinBatchSize) {
         checkArgument(spInsertBatchSize > 0, "The SP insert batch size '" + spInsertBatchSize + "' must be greater than 0.");
         checkArgument(joinBatchSize > 0, "The Join batch size '" + joinBatchSize + "' must be greater than 0.");
         this.spInsertBatchSize = spInsertBatchSize;
         this.joinBatchSize = joinBatchSize;
     }


     /**
      * Tells the Fluo PCJ Updater application to maintain a new PCJ. This method
      * creates the FluoQuery (metadata) inside of Fluo so that results can be incrementally generated
      * inside of Fluo.  This method assumes that the user will export the results to Kafka
      * according to the Kafka {@link ExportStrategy}.
      *
      * @param sparql - sparql query String to be registered with Fluo
      * @param strategies - ExportStrategies used to specify how final results will be handled
      * @param fluo - A connection to the Fluo application that updates the PCJ index. (not null)
      * @return The metadata that was written to the Fluo application for the PCJ.
      * @throws MalformedQueryException The SPARQL query stored for the {@code pcjId} is malformed.
      * @throws UnsupportedQueryException
      * @throws PcjException The PCJ Metadata for {@code pcjId} could not be read from {@code pcjStorage}.
      */
     public FluoQuery createPcj(String sparql, FluoClient fluo) throws MalformedQueryException, UnsupportedQueryException {
         Preconditions.checkNotNull(sparql);
         Preconditions.checkNotNull(fluo);

         String pcjId = FluoQueryUtils.createNewPcjId();
         return createPcj(pcjId, sparql, Sets.newHashSet(ExportStrategy.KAFKA), fluo);
     }

     /**
      * Tells the Fluo PCJ Updater application to maintain a new PCJ.  This method provides
      * no guarantees that a PCJ with the given pcjId exists outside of Fluo. This method merely
      * creates the FluoQuery (metadata) inside of Fluo so that results can be incrementally generated
      * inside of Fluo.  Results are exported according to the Set of {@link ExportStrategy} enums.  If
      * the Rya ExportStrategy is specified, care should be taken to verify that the PCJ table exists.
      *
      * @param pcjId - Identifies the PCJ that will be updated by the Fluo app. (not null)
      * @param sparql - sparql query String to be registered with Fluo
      * @param strategies - ExportStrategies used to specify how final results will be handled
      * @param fluo - A connection to the Fluo application that updates the PCJ index. (not null)
      * @return The metadata that was written to the Fluo application for the PCJ.
      * @throws UnsupportedQueryException
      * @throws MalformedQueryException
      */
     public FluoQuery createPcj(
             final String pcjId,
             final String sparql,
             final Set<ExportStrategy> strategies,
             final FluoClient fluo) throws MalformedQueryException, UnsupportedQueryException {
         requireNonNull(pcjId);
         requireNonNull(sparql);
         requireNonNull(strategies);
         requireNonNull(fluo);

         FluoQuery fluoQuery = makeFluoQuery(sparql, pcjId, strategies);
         writeFluoQuery(fluo, fluoQuery, pcjId);

         return fluoQuery;
     }

     /**
      * Tells the Fluo PCJ Updater application to maintain a new PCJ.  The method takes in an
      * instance of {@link PrecomputedJoinStorage} to verify that a PCJ with the given pcjId exists.
      * Results are exported to a PCJ table with the provided pcjId according to the Rya
      * {@link ExportStrategy}.
      *
      * @param pcjId - Identifies the PCJ that will be updated by the Fluo app. (not null)
      * @param pcjStorage - Provides access to the PCJ index. (not null)
      * @param fluo - A connection to the Fluo application that updates the PCJ index. (not null)
      * @return The metadata that was written to the Fluo application for the PCJ.
      * @throws MalformedQueryException The SPARQL query stored for the {@code pcjId} is malformed.
      * @throws PcjException The PCJ Metadata for {@code pcjId} could not be read from {@code pcjStorage}.
      * @throws UnsupportedQueryException
      */
     public FluoQuery createPcj(
             final String pcjId,
             final PrecomputedJoinStorage pcjStorage,
             final FluoClient fluo) throws MalformedQueryException, PcjException, UnsupportedQueryException {
         requireNonNull(pcjId);
         requireNonNull(pcjStorage);
         requireNonNull(fluo);

         // Parse the query's structure for the metadata that will be written to fluo.
         final PcjMetadata pcjMetadata = pcjStorage.getPcjMetadata(pcjId);
         final String sparql = pcjMetadata.getSparql();
         return createPcj(pcjId, sparql, Sets.newHashSet(ExportStrategy.RYA), fluo);
     }

     private FluoQuery makeFluoQuery(String sparql, String pcjId, Set<ExportStrategy> strategies) throws MalformedQueryException, UnsupportedQueryException {

         String queryId = NodeType.generateNewIdForType(NodeType.QUERY, pcjId);

         SparqlFluoQueryBuilder builder = new SparqlFluoQueryBuilder()
                 .setExportStrategies(strategies)
                 .setFluoQueryId(queryId)
                 .setSparql(sparql)
                 .setJoinBatchSize(joinBatchSize);

         FluoQuery query = builder.build();

         if(query.getQueryType() == QueryType.PERIODIC && !Sets.newHashSet(ExportStrategy.PERIODIC).containsAll(strategies)) {
             throw new UnsupportedQueryException("Periodic Queries must only utilize the PeriodicExport or the NoOpExport ExportStrategy.");
         }

         if(query.getQueryType() != QueryType.PERIODIC && strategies.contains(ExportStrategy.PERIODIC)) {
             throw new UnsupportedQueryException("Only Periodic Queries can utilize the PeriodicExport ExportStrategy.");
         }

         return query;
     }

     private void writeFluoQuery(FluoClient fluo, FluoQuery fluoQuery, String pcjId) {
         try (Transaction tx = fluo.newTransaction()) {
             // Write the query's structure to Fluo.
             new FluoQueryMetadataDAO().write(tx, fluoQuery);

             // Flush the changes to Fluo.
             tx.commit();
         }
     }

     /**
      * Tells the Fluo PCJ Updater application to maintain a new PCJ.
      * <p>
      * This call scans Rya for Statement Pattern matches and inserts them into
      * the Fluo application. It is assumed that results for any query registered
      * using this method will be exported to Kafka according to the Kafka {@link ExportStrategy}.
      *
      * @param sparql - sparql query that will registered with Fluo. (not null)
      * @param fluo - A connection to the Fluo application that updates the PCJ index. (not null)
      * @param accumulo - Accumulo connector for connecting with Accumulo
      * @param ryaInstance - Name of Rya instance to connect to
      * @return The Fluo application's Query ID of the query that was created.
      * @throws MalformedQueryException The SPARQL query stored for the {@code pcjId} is malformed.
      * @throws PcjException The PCJ Metadata for {@code pcjId} could not be read from {@code pcjStorage}.
      * @throws RyaDAOException Historic PCJ results could not be loaded because of a problem with {@code rya}.
      * @throws UnsupportedQueryException
      */
     public String withRyaIntegration(
             final String sparql,
             final FluoClient fluo,
             final Connector accumulo,
             final String ryaInstance ) throws MalformedQueryException, PcjException, RyaDAOException, UnsupportedQueryException {
         requireNonNull(sparql);
         requireNonNull(fluo);
         requireNonNull(accumulo);
         requireNonNull(ryaInstance);


         // Write the SPARQL query's structure to the Fluo Application.
         final FluoQuery fluoQuery = createPcj(sparql, fluo);
         //import results already ingested into Rya that match query
         importHistoricResultsIntoFluo(fluo, fluoQuery, accumulo, ryaInstance);
         // return queryId to the caller for later monitoring from the export.
         return fluoQuery.getQueryMetadata().getNodeId();
     }


     /**
      * Tells the Fluo PCJ Updater application to maintain a new PCJ.
      * <p>
      * This call scans Rya for Statement Pattern matches and inserts them into
      * the Fluo application. This method does not verify that a PcjTable with the
      * the given pcjId actually exists, so one should verify that the table exists before
      * using the Rya ExportStrategy. Results will be exported according to the Set of
      * {@link ExportStrategy} enums.
      *
      * @param pcjId - Identifies the PCJ that will be updated by the Fluo app. (not null)
      * @param sparql - sparql query that will registered with Fluo. (not null)
      * @param strategies - ExportStrategies used to specify how final results will be handled
      * @param fluo - A connection to the Fluo application that updates the PCJ index. (not null)
      * @param accumulo - Accumulo connector for connecting with Accumulo
      * @param ryaInstance - name of Rya instance to connect to
      * @return FluoQuery containing the metadata for the newly registered SPARQL query
      * @throws MalformedQueryException The SPARQL query stored for the {@code pcjId} is malformed.
      * @throws PcjException The PCJ Metadata for {@code pcjId} could not be read from {@code pcjStorage}.
      * @throws RyaDAOException Historic PCJ results could not be loaded because of a problem with {@code rya}.
      * @throws UnsupportedQueryException
      */
     public FluoQuery withRyaIntegration(
             final String pcjId,
             final String sparql,
             final Set<ExportStrategy> strategies,
             final FluoClient fluo,
             final Connector accumulo,
             final String ryaInstance ) throws MalformedQueryException, PcjException, RyaDAOException, UnsupportedQueryException {
         requireNonNull(pcjId);
         requireNonNull(sparql);
         requireNonNull(fluo);
         requireNonNull(accumulo);
         requireNonNull(ryaInstance);


         // Write the SPARQL query's structure to the Fluo Application.
         final FluoQuery fluoQuery = createPcj(pcjId, sparql, strategies, fluo);
         //import results already ingested into Rya that match query
         importHistoricResultsIntoFluo(fluo, fluoQuery, accumulo, ryaInstance);
         // return queryId to the caller for later monitoring from the export.
         return fluoQuery;
     }

     /**
      * Tells the Fluo PCJ Updater application to maintain a new PCJ.
      * <p>
      * This call scans Rya for Statement Pattern matches and inserts them into
      * the Fluo application. The Fluo application will then maintain the intermediate
      * results as new triples are inserted and export any new query results to the
      * {@code pcjId} within the provided {@code pcjStorage}.  This method requires that a
      * PCJ table already exist for the query corresponding to the pcjId.  By default, results will be exported
      * to this table according to the Rya {@link ExportStrategy}.
      *
      * @param pcjId - Identifies the PCJ that will be updated by the Fluo app. (not null)
      * @param pcjStorage - Provides access to the PCJ index. (not null)
      * @param fluo - A connection to the Fluo application that updates the PCJ index. (not null)
      * @param accumulo - Accumuo connector for connecting to Accumulo
      * @param ryaInstance - name of Rya instance to connect to
      * @return FluoQuery containing the metadata for the newly registered SPARQL query
      * @throws MalformedQueryException The SPARQL query stored for the {@code pcjId} is malformed.
      * @throws PcjException The PCJ Metadata for {@code pcjId} could not be read from {@code pcjStorage}.
      * @throws RyaDAOException Historic PCJ results could not be loaded because of a problem with {@code rya}.
      * @throws UnsupportedQueryException
      */
     public FluoQuery withRyaIntegration(
             final String pcjId,
             final PrecomputedJoinStorage pcjStorage,
             final FluoClient fluo,
             final Connector accumulo,
             final String ryaInstance ) throws MalformedQueryException, PcjException, RyaDAOException, UnsupportedQueryException {
         requireNonNull(pcjId);
         requireNonNull(pcjStorage);
         requireNonNull(fluo);
         requireNonNull(accumulo);
         requireNonNull(ryaInstance);

         // Parse the query's structure for the metadata that will be written to fluo.
         final PcjMetadata pcjMetadata = pcjStorage.getPcjMetadata(pcjId);
         final String sparql = pcjMetadata.getSparql();

         return withRyaIntegration(pcjId, sparql, Sets.newHashSet(ExportStrategy.RYA), fluo, accumulo, ryaInstance);
     }


     private void importHistoricResultsIntoFluo(FluoClient fluo, FluoQuery fluoQuery, Connector accumulo, String ryaInstance)
             throws RyaDAOException {
         // Reuse the same set object while performing batch inserts.
         final Set<RyaStatement> queryBatch = new HashSet<>();

         // Iterate through each of the statement patterns and insert their
         // historic matches into Fluo.
         for (final StatementPatternMetadata patternMetadata : fluoQuery.getStatementPatternMetadata()) {
             // Get an iterator over all of the binding sets that match the
             // statement pattern.
             final StatementPattern pattern = FluoStringConverter.toStatementPattern(patternMetadata.getStatementPattern());
             queryBatch.add(spToRyaStatement(pattern));
         }

         // Create AccumuloRyaQueryEngine to query for historic results
         final AccumuloRdfConfiguration conf = new AccumuloRdfConfiguration();
         conf.setTablePrefix(ryaInstance);
         conf.setAuths(getAuths(accumulo));

         try (final AccumuloRyaQueryEngine queryEngine = new AccumuloRyaQueryEngine(accumulo, conf);
                 CloseableIterable<RyaStatement> queryIterable = queryEngine.query(new BatchRyaQuery(queryBatch))) {
             final Set<RyaStatement> triplesBatch = new HashSet<>();

             // Insert batches of the binding sets into Fluo.
             for (final RyaStatement ryaStatement : queryIterable) {
                 if (triplesBatch.size() == spInsertBatchSize) {
                     writeBatch(fluo, triplesBatch);
                     triplesBatch.clear();
                 }

                 triplesBatch.add(ryaStatement);
             }

             if (!triplesBatch.isEmpty()) {
                 writeBatch(fluo, triplesBatch);
                 triplesBatch.clear();
             }
         } catch (final IOException e) {
             log.warn("Ignoring IOException thrown while closing the AccumuloRyaQueryEngine used by CreatePCJ.", e);
         }
     }

     private static void writeBatch(final FluoClient fluo, final Set<RyaStatement> batch) {
         checkNotNull(fluo);
         checkNotNull(batch);
         new InsertTriples().insert(fluo, batch);
     }

     private static RyaStatement spToRyaStatement(final StatementPattern sp) {
         final Value subjVal = sp.getSubjectVar().getValue();
         final Value predVal = sp.getPredicateVar().getValue();
         final Value objVal = sp.getObjectVar().getValue();

         RyaURI subjURI = null;
         RyaURI predURI = null;
         RyaType objType = null;

         if(subjVal != null) {
             if(!(subjVal instanceof Resource)) {
                 throw new AssertionError("Subject must be a Resource.");
             }
             subjURI = RdfToRyaConversions.convertResource((Resource) subjVal);
         }

         if (predVal != null) {
             if(!(predVal instanceof IRI)) {
                 throw new AssertionError("Predicate must be a URI.");
             }
             predURI = RdfToRyaConversions.convertURI((IRI) predVal);
         }

         if (objVal != null ) {
             objType = RdfToRyaConversions.convertValue(objVal);
         }

         return new RyaStatement(subjURI, predURI, objType);
     }

     private String[] getAuths(final Connector accumulo) {
         Authorizations auths;
         try {
             auths = accumulo.securityOperations().getUserAuthorizations(accumulo.whoami());
             final List<byte[]> authList = auths.getAuthorizations();
             final String[] authArray = new String[authList.size()];
             for(int i = 0; i < authList.size(); i++){
                 authArray[i] = new String(authList.get(i), "UTF-8");
             }
             return authArray;
         } catch (AccumuloException | AccumuloSecurityException | UnsupportedEncodingException e) {
             throw new RuntimeException("Cannot read authorizations for user: " + accumulo.whoami());
         }
     }
 }
	/*
	* 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.api;

	import static com.google.common.base.Preconditions.checkArgument;
	import static com.google.common.base.Preconditions.checkNotNull;
	import static java.util.Objects.requireNonNull;

	import java.io.IOException;
	import java.io.UnsupportedEncodingException;
	import java.util.HashSet;
	import java.util.List;
	import java.util.Set;

	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.security.Authorizations;
	import org.apache.fluo.api.client.FluoClient;
	import org.apache.fluo.api.client.Transaction;
	import org.apache.log4j.Logger;
	import org.apache.rya.accumulo.AccumuloRdfConfiguration;
	import org.apache.rya.accumulo.query.AccumuloRyaQueryEngine;
	import org.apache.rya.api.client.CreatePCJ.ExportStrategy;
	import org.apache.rya.api.client.CreatePCJ.QueryType;
	import org.apache.rya.api.domain.RyaStatement;
	import org.apache.rya.api.domain.RyaType;
	import org.apache.rya.api.domain.RyaURI;
	import org.apache.rya.api.persist.RyaDAOException;
	import org.apache.rya.api.persist.query.BatchRyaQuery;
	import org.apache.rya.api.resolver.RdfToRyaConversions;
	import org.apache.rya.indexing.pcj.fluo.app.FluoStringConverter;
	import org.apache.rya.indexing.pcj.fluo.app.NodeType;
	import org.apache.rya.indexing.pcj.fluo.app.query.FluoQuery;
	import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryMetadataDAO;
	import org.apache.rya.indexing.pcj.fluo.app.query.SparqlFluoQueryBuilder;
	import org.apache.rya.indexing.pcj.fluo.app.query.StatementPatternMetadata;
	import org.apache.rya.indexing.pcj.fluo.app.query.UnsupportedQueryException;
	import org.apache.rya.indexing.pcj.fluo.app.util.FluoQueryUtils;
	import org.apache.rya.indexing.pcj.storage.PcjException;
	import org.apache.rya.indexing.pcj.storage.PcjMetadata;
	import org.apache.rya.indexing.pcj.storage.PrecomputedJoinStorage;
	import org.calrissian.mango.collect.CloseableIterable;
	import org.eclipse.rdf4j.model.IRI;
	import org.eclipse.rdf4j.model.Resource;
	import org.eclipse.rdf4j.model.Value;
	import org.eclipse.rdf4j.query.MalformedQueryException;
	import org.eclipse.rdf4j.query.algebra.StatementPattern;

	import com.google.common.base.Preconditions;
	import com.google.common.collect.Sets;

	import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
	import edu.umd.cs.findbugs.annotations.NonNull;

	/**
	* Sets up a new Pre Computed Join (PCJ) in Fluo from a SPARQL query.
	* <p>
	* This is a two phase process.
	* <ol>
	* <li>Setup metadata about each node of the query using a single Fluo transaction. </li>
	* <li>Scan Rya for binding sets that match each Statement Pattern from the query
	* and use a separate Fluo transaction for each batch that is inserted. This
	* ensure historic triples will be included in the query's results.</li>
	* </ol>
	* After the first step is finished, any new Triples that are added to the Fluo
	* application will be matched against statement patterns, the final results
	* will percolate to the top of the query, and those results will be exported to
	* Rya's query system.
	*/
	@DefaultAnnotation(NonNull.class)
	public class CreateFluoPcj {
	private static final Logger log = Logger.getLogger(CreateFluoPcj.class);

	/**
	* The default Statement Pattern batch insert size is 1000.
	*/
	private static final int DEFAULT_SP_INSERT_BATCH_SIZE = 1000;
	/**
	* The default Join batch size is 5000.
	*/
	private static final int DEFAULT_JOIN_BATCH_SIZE = 5000;

	/**
	* The maximum number of binding sets that will be inserted into each Statement
	* Pattern's result set per Fluo transaction.
	*/
	private final int spInsertBatchSize;

	/**
	* The maximum number of join results that will be processed per transaction.
	*/
	private final int joinBatchSize;
	/**
	* Constructs an instance of {@link CreateFluoPcj} that uses
	* {@link #DEFAULT_SP_INSERT_BATCH_SIZE} as the default batch insert size.
	*/
	public CreateFluoPcj() {
	this(DEFAULT_SP_INSERT_BATCH_SIZE, DEFAULT_JOIN_BATCH_SIZE);
	}

	/**
	* Constructs an instance of {@link CreateFluoPcj}.
	*
	* @param spInsertBatchSize - The maximum number of binding sets that will be
	* inserted into each Statement Pattern's result set per Fluo transaction.
	*/
	public CreateFluoPcj(final int spInsertBatchSize, final int joinBatchSize) {
	checkArgument(spInsertBatchSize > 0, "The SP insert batch size '" + spInsertBatchSize + "' must be greater than 0.");
	checkArgument(joinBatchSize > 0, "The Join batch size '" + joinBatchSize + "' must be greater than 0.");
	this.spInsertBatchSize = spInsertBatchSize;
	this.joinBatchSize = joinBatchSize;
	}


	/**
	* Tells the Fluo PCJ Updater application to maintain a new PCJ. This method
	* creates the FluoQuery (metadata) inside of Fluo so that results can be incrementally generated
	* inside of Fluo. This method assumes that the user will export the results to Kafka
	* according to the Kafka {@link ExportStrategy}.
	*
	* @param sparql - sparql query String to be registered with Fluo
	* @param strategies - ExportStrategies used to specify how final results will be handled
	* @param fluo - A connection to the Fluo application that updates the PCJ index. (not null)
	* @return The metadata that was written to the Fluo application for the PCJ.
	* @throws MalformedQueryException The SPARQL query stored for the {@code pcjId} is malformed.
	* @throws UnsupportedQueryException
	* @throws PcjException The PCJ Metadata for {@code pcjId} could not be read from {@code pcjStorage}.
	*/
	public FluoQuery createPcj(String sparql, FluoClient fluo) throws MalformedQueryException, UnsupportedQueryException {
	Preconditions.checkNotNull(sparql);
	Preconditions.checkNotNull(fluo);

	String pcjId = FluoQueryUtils.createNewPcjId();
	return createPcj(pcjId, sparql, Sets.newHashSet(ExportStrategy.KAFKA), fluo);
	}

	/**
	* Tells the Fluo PCJ Updater application to maintain a new PCJ. This method provides
	* no guarantees that a PCJ with the given pcjId exists outside of Fluo. This method merely
	* creates the FluoQuery (metadata) inside of Fluo so that results can be incrementally generated
	* inside of Fluo. Results are exported according to the Set of {@link ExportStrategy} enums. If
	* the Rya ExportStrategy is specified, care should be taken to verify that the PCJ table exists.
	*
	* @param pcjId - Identifies the PCJ that will be updated by the Fluo app. (not null)
	* @param sparql - sparql query String to be registered with Fluo
	* @param strategies - ExportStrategies used to specify how final results will be handled
	* @param fluo - A connection to the Fluo application that updates the PCJ index. (not null)
	* @return The metadata that was written to the Fluo application for the PCJ.
	* @throws UnsupportedQueryException
	* @throws MalformedQueryException
	*/
	public FluoQuery createPcj(
	final String pcjId,
	final String sparql,
	final Set<ExportStrategy> strategies,
	final FluoClient fluo) throws MalformedQueryException, UnsupportedQueryException {
	requireNonNull(pcjId);
	requireNonNull(sparql);
	requireNonNull(strategies);
	requireNonNull(fluo);

	FluoQuery fluoQuery = makeFluoQuery(sparql, pcjId, strategies);
	writeFluoQuery(fluo, fluoQuery, pcjId);

	return fluoQuery;
	}

	/**
	* Tells the Fluo PCJ Updater application to maintain a new PCJ. The method takes in an
	* instance of {@link PrecomputedJoinStorage} to verify that a PCJ with the given pcjId exists.
	* Results are exported to a PCJ table with the provided pcjId according to the Rya
	* {@link ExportStrategy}.
	*
	* @param pcjId - Identifies the PCJ that will be updated by the Fluo app. (not null)
	* @param pcjStorage - Provides access to the PCJ index. (not null)
	* @param fluo - A connection to the Fluo application that updates the PCJ index. (not null)
	* @return The metadata that was written to the Fluo application for the PCJ.
	* @throws MalformedQueryException The SPARQL query stored for the {@code pcjId} is malformed.
	* @throws PcjException The PCJ Metadata for {@code pcjId} could not be read from {@code pcjStorage}.
	* @throws UnsupportedQueryException
	*/
	public FluoQuery createPcj(
	final String pcjId,
	final PrecomputedJoinStorage pcjStorage,
	final FluoClient fluo) throws MalformedQueryException, PcjException, UnsupportedQueryException {
	requireNonNull(pcjId);
	requireNonNull(pcjStorage);
	requireNonNull(fluo);

	// Parse the query's structure for the metadata that will be written to fluo.
	final PcjMetadata pcjMetadata = pcjStorage.getPcjMetadata(pcjId);
	final String sparql = pcjMetadata.getSparql();
	return createPcj(pcjId, sparql, Sets.newHashSet(ExportStrategy.RYA), fluo);
	}

	private FluoQuery makeFluoQuery(String sparql, String pcjId, Set<ExportStrategy> strategies) throws MalformedQueryException, UnsupportedQueryException {

	String queryId = NodeType.generateNewIdForType(NodeType.QUERY, pcjId);

	SparqlFluoQueryBuilder builder = new SparqlFluoQueryBuilder()
	.setExportStrategies(strategies)
	.setFluoQueryId(queryId)
	.setSparql(sparql)
	.setJoinBatchSize(joinBatchSize);

	FluoQuery query = builder.build();

	if(query.getQueryType() == QueryType.PERIODIC && !Sets.newHashSet(ExportStrategy.PERIODIC).containsAll(strategies)) {
	throw new UnsupportedQueryException("Periodic Queries must only utilize the PeriodicExport or the NoOpExport ExportStrategy.");
	}

	if(query.getQueryType() != QueryType.PERIODIC && strategies.contains(ExportStrategy.PERIODIC)) {
	throw new UnsupportedQueryException("Only Periodic Queries can utilize the PeriodicExport ExportStrategy.");
	}

	return query;
	}

	private void writeFluoQuery(FluoClient fluo, FluoQuery fluoQuery, String pcjId) {
	try (Transaction tx = fluo.newTransaction()) {
	// Write the query's structure to Fluo.
	new FluoQueryMetadataDAO().write(tx, fluoQuery);

	// Flush the changes to Fluo.
	tx.commit();
	}
	}

	/**
	* Tells the Fluo PCJ Updater application to maintain a new PCJ.
	* <p>
	* This call scans Rya for Statement Pattern matches and inserts them into
	* the Fluo application. It is assumed that results for any query registered
	* using this method will be exported to Kafka according to the Kafka {@link ExportStrategy}.
	*
	* @param sparql - sparql query that will registered with Fluo. (not null)
	* @param fluo - A connection to the Fluo application that updates the PCJ index. (not null)
	* @param accumulo - Accumulo connector for connecting with Accumulo
	* @param ryaInstance - Name of Rya instance to connect to
	* @return The Fluo application's Query ID of the query that was created.
	* @throws MalformedQueryException The SPARQL query stored for the {@code pcjId} is malformed.
	* @throws PcjException The PCJ Metadata for {@code pcjId} could not be read from {@code pcjStorage}.
	* @throws RyaDAOException Historic PCJ results could not be loaded because of a problem with {@code rya}.
	* @throws UnsupportedQueryException
	*/
	public String withRyaIntegration(
	final String sparql,
	final FluoClient fluo,
	final Connector accumulo,
	final String ryaInstance ) throws MalformedQueryException, PcjException, RyaDAOException, UnsupportedQueryException {
	requireNonNull(sparql);
	requireNonNull(fluo);
	requireNonNull(accumulo);
	requireNonNull(ryaInstance);


	// Write the SPARQL query's structure to the Fluo Application.
	final FluoQuery fluoQuery = createPcj(sparql, fluo);
	//import results already ingested into Rya that match query
	importHistoricResultsIntoFluo(fluo, fluoQuery, accumulo, ryaInstance);
	// return queryId to the caller for later monitoring from the export.
	return fluoQuery.getQueryMetadata().getNodeId();
	}


	/**
	* Tells the Fluo PCJ Updater application to maintain a new PCJ.
	* <p>
	* This call scans Rya for Statement Pattern matches and inserts them into
	* the Fluo application. This method does not verify that a PcjTable with the
	* the given pcjId actually exists, so one should verify that the table exists before
	* using the Rya ExportStrategy. Results will be exported according to the Set of
	* {@link ExportStrategy} enums.
	*
	* @param pcjId - Identifies the PCJ that will be updated by the Fluo app. (not null)
	* @param sparql - sparql query that will registered with Fluo. (not null)
	* @param strategies - ExportStrategies used to specify how final results will be handled
	* @param fluo - A connection to the Fluo application that updates the PCJ index. (not null)
	* @param accumulo - Accumulo connector for connecting with Accumulo
	* @param ryaInstance - name of Rya instance to connect to
	* @return FluoQuery containing the metadata for the newly registered SPARQL query
	* @throws MalformedQueryException The SPARQL query stored for the {@code pcjId} is malformed.
	* @throws PcjException The PCJ Metadata for {@code pcjId} could not be read from {@code pcjStorage}.
	* @throws RyaDAOException Historic PCJ results could not be loaded because of a problem with {@code rya}.
	* @throws UnsupportedQueryException
	*/
	public FluoQuery withRyaIntegration(
	final String pcjId,
	final String sparql,
	final Set<ExportStrategy> strategies,
	final FluoClient fluo,
	final Connector accumulo,
	final String ryaInstance ) throws MalformedQueryException, PcjException, RyaDAOException, UnsupportedQueryException {
	requireNonNull(pcjId);
	requireNonNull(sparql);
	requireNonNull(fluo);
	requireNonNull(accumulo);
	requireNonNull(ryaInstance);


	// Write the SPARQL query's structure to the Fluo Application.
	final FluoQuery fluoQuery = createPcj(pcjId, sparql, strategies, fluo);
	//import results already ingested into Rya that match query
	importHistoricResultsIntoFluo(fluo, fluoQuery, accumulo, ryaInstance);
	// return queryId to the caller for later monitoring from the export.
	return fluoQuery;
	}

	/**
	* Tells the Fluo PCJ Updater application to maintain a new PCJ.
	* <p>
	* This call scans Rya for Statement Pattern matches and inserts them into
	* the Fluo application. The Fluo application will then maintain the intermediate
	* results as new triples are inserted and export any new query results to the
	* {@code pcjId} within the provided {@code pcjStorage}. This method requires that a
	* PCJ table already exist for the query corresponding to the pcjId. By default, results will be exported
	* to this table according to the Rya {@link ExportStrategy}.
	*
	* @param pcjId - Identifies the PCJ that will be updated by the Fluo app. (not null)
	* @param pcjStorage - Provides access to the PCJ index. (not null)
	* @param fluo - A connection to the Fluo application that updates the PCJ index. (not null)
	* @param accumulo - Accumuo connector for connecting to Accumulo
	* @param ryaInstance - name of Rya instance to connect to
	* @return FluoQuery containing the metadata for the newly registered SPARQL query
	* @throws MalformedQueryException The SPARQL query stored for the {@code pcjId} is malformed.
	* @throws PcjException The PCJ Metadata for {@code pcjId} could not be read from {@code pcjStorage}.
	* @throws RyaDAOException Historic PCJ results could not be loaded because of a problem with {@code rya}.
	* @throws UnsupportedQueryException
	*/
	public FluoQuery withRyaIntegration(
	final String pcjId,
	final PrecomputedJoinStorage pcjStorage,
	final FluoClient fluo,
	final Connector accumulo,
	final String ryaInstance ) throws MalformedQueryException, PcjException, RyaDAOException, UnsupportedQueryException {
	requireNonNull(pcjId);
	requireNonNull(pcjStorage);
	requireNonNull(fluo);
	requireNonNull(accumulo);
	requireNonNull(ryaInstance);

	// Parse the query's structure for the metadata that will be written to fluo.
	final PcjMetadata pcjMetadata = pcjStorage.getPcjMetadata(pcjId);
	final String sparql = pcjMetadata.getSparql();

	return withRyaIntegration(pcjId, sparql, Sets.newHashSet(ExportStrategy.RYA), fluo, accumulo, ryaInstance);
	}



	private void importHistoricResultsIntoFluo(FluoClient fluo, FluoQuery fluoQuery, Connector accumulo, String ryaInstance)
	throws RyaDAOException {
	// Reuse the same set object while performing batch inserts.
	final Set<RyaStatement> queryBatch = new HashSet<>();

	// Iterate through each of the statement patterns and insert their
	// historic matches into Fluo.
	for (final StatementPatternMetadata patternMetadata : fluoQuery.getStatementPatternMetadata()) {
	// Get an iterator over all of the binding sets that match the
	// statement pattern.
	final StatementPattern pattern = FluoStringConverter.toStatementPattern(patternMetadata.getStatementPattern());
	queryBatch.add(spToRyaStatement(pattern));
	}

	// Create AccumuloRyaQueryEngine to query for historic results
	final AccumuloRdfConfiguration conf = new AccumuloRdfConfiguration();
	conf.setTablePrefix(ryaInstance);
	conf.setAuths(getAuths(accumulo));

	try (final AccumuloRyaQueryEngine queryEngine = new AccumuloRyaQueryEngine(accumulo, conf);
	CloseableIterable<RyaStatement> queryIterable = queryEngine.query(new BatchRyaQuery(queryBatch))) {
	final Set<RyaStatement> triplesBatch = new HashSet<>();

	// Insert batches of the binding sets into Fluo.
	for (final RyaStatement ryaStatement : queryIterable) {
	if (triplesBatch.size() == spInsertBatchSize) {
	writeBatch(fluo, triplesBatch);
	triplesBatch.clear();
	}

	triplesBatch.add(ryaStatement);
	}

	if (!triplesBatch.isEmpty()) {
	writeBatch(fluo, triplesBatch);
	triplesBatch.clear();
	}
	} catch (final IOException e) {
	log.warn("Ignoring IOException thrown while closing the AccumuloRyaQueryEngine used by CreatePCJ.", e);
	}
	}

	private static void writeBatch(final FluoClient fluo, final Set<RyaStatement> batch) {
	checkNotNull(fluo);
	checkNotNull(batch);
	new InsertTriples().insert(fluo, batch);
	}

	private static RyaStatement spToRyaStatement(final StatementPattern sp) {
	final Value subjVal = sp.getSubjectVar().getValue();
	final Value predVal = sp.getPredicateVar().getValue();
	final Value objVal = sp.getObjectVar().getValue();

	RyaURI subjURI = null;
	RyaURI predURI = null;
	RyaType objType = null;

	if(subjVal != null) {
	if(!(subjVal instanceof Resource)) {
	throw new AssertionError("Subject must be a Resource.");
	}
	subjURI = RdfToRyaConversions.convertResource((Resource) subjVal);
	}

	if (predVal != null) {
	if(!(predVal instanceof IRI)) {
	throw new AssertionError("Predicate must be a URI.");
	}
	predURI = RdfToRyaConversions.convertURI((IRI) predVal);
	}

	if (objVal != null ) {
	objType = RdfToRyaConversions.convertValue(objVal);
	}

	return new RyaStatement(subjURI, predURI, objType);
	}

	private String[] getAuths(final Connector accumulo) {
	Authorizations auths;
	try {
	auths = accumulo.securityOperations().getUserAuthorizations(accumulo.whoami());
	final List<byte[]> authList = auths.getAuthorizations();
	final String[] authArray = new String[authList.size()];
	for(int i = 0; i < authList.size(); i++){
	authArray[i] = new String(authList.get(i), "UTF-8");
	}
	return authArray;
	} catch (AccumuloException \| AccumuloSecurityException \| UnsupportedEncodingException e) {
	throw new RuntimeException("Cannot read authorizations for user: " + accumulo.whoami());
	}
	}
	}