You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rya.apache.org by ca...@apache.org on 2017/08/02 21:02:01 UTC
[8/9] incubator-rya git commit: RYA-280-Periodic Query Service.
Closes #177.
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/integration/AccumuloPeriodicQueryResultStorageIT.java
----------------------------------------------------------------------
diff --git a/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/integration/AccumuloPeriodicQueryResultStorageIT.java b/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/integration/AccumuloPeriodicQueryResultStorageIT.java
new file mode 100644
index 0000000..1eafc00
--- /dev/null
+++ b/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/integration/AccumuloPeriodicQueryResultStorageIT.java
@@ -0,0 +1,271 @@
+/*
+ * 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.storage.accumulo.integration;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.rya.accumulo.AccumuloITBase;
+import org.apache.rya.indexing.pcj.storage.PeriodicQueryResultStorage;
+import org.apache.rya.indexing.pcj.storage.PeriodicQueryStorageException;
+import org.apache.rya.indexing.pcj.storage.PeriodicQueryStorageMetadata;
+import org.apache.rya.indexing.pcj.storage.PrecomputedJoinStorage.CloseableIterator;
+import org.apache.rya.indexing.pcj.storage.accumulo.AccumuloPeriodicQueryResultStorage;
+import org.apache.rya.indexing.pcj.storage.accumulo.PeriodicQueryTableNameFactory;
+import org.apache.rya.indexing.pcj.storage.accumulo.VariableOrder;
+import org.apache.rya.indexing.pcj.storage.accumulo.VisibilityBindingSet;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.openrdf.model.ValueFactory;
+import org.openrdf.model.impl.ValueFactoryImpl;
+import org.openrdf.model.vocabulary.XMLSchema;
+import org.openrdf.query.BindingSet;
+import org.openrdf.query.algebra.evaluation.QueryBindingSet;
+import org.openrdf.query.impl.MapBindingSet;
+
+public class AccumuloPeriodicQueryResultStorageIT extends AccumuloITBase {
+
+ private PeriodicQueryResultStorage periodicStorage;
+ private static final String RYA = "rya_";
+ private static final PeriodicQueryTableNameFactory nameFactory = new PeriodicQueryTableNameFactory();
+ private static final ValueFactory vf = new ValueFactoryImpl();
+
+ @Before
+ public void init() throws AccumuloException, AccumuloSecurityException {
+ super.getConnector().securityOperations().changeUserAuthorizations("root", new Authorizations("U"));
+ periodicStorage = new AccumuloPeriodicQueryResultStorage(super.getConnector(), RYA);
+ }
+
+
+ @Test
+ public void testCreateAndMeta() throws PeriodicQueryStorageException {
+
+ String sparql = "select ?x where { ?x <urn:pred> ?y.}";
+ VariableOrder varOrder = new VariableOrder("periodicBinId", "x");
+ PeriodicQueryStorageMetadata expectedMeta = new PeriodicQueryStorageMetadata(sparql, varOrder);
+
+ String id = periodicStorage.createPeriodicQuery(sparql);
+ Assert.assertEquals(expectedMeta, periodicStorage.getPeriodicQueryMetadata(id));
+ Assert.assertEquals(Arrays.asList(nameFactory.makeTableName(RYA, id)), periodicStorage.listPeriodicTables());
+ periodicStorage.deletePeriodicQuery(id);
+ }
+
+
+ @Test
+ public void testAddListDelete() throws Exception {
+
+ String sparql = "select ?x where { ?x <urn:pred> ?y.}";
+ String id = periodicStorage.createPeriodicQuery(sparql);
+
+ Set<BindingSet> expected = new HashSet<>();
+ Set<VisibilityBindingSet> storageSet = new HashSet<>();
+
+ //add result matching user's visibility
+ QueryBindingSet bs = new QueryBindingSet();
+ bs.addBinding("periodicBinId", vf.createLiteral(1L));
+ bs.addBinding("x",vf.createURI("uri:uri123"));
+ expected.add(bs);
+ storageSet.add(new VisibilityBindingSet(bs,"U"));
+
+ //add result with different visibility that is not expected
+ bs = new QueryBindingSet();
+ bs.addBinding("periodicBinId", vf.createLiteral(1L));
+ bs.addBinding("x",vf.createURI("uri:uri456"));
+ storageSet.add(new VisibilityBindingSet(bs,"V"));
+
+ periodicStorage.addPeriodicQueryResults(id, storageSet);
+
+ Set<BindingSet> actual = new HashSet<>();
+ try(CloseableIterator<BindingSet> iter = periodicStorage.listResults(id, Optional.of(1L))) {
+ iter.forEachRemaining(x -> actual.add(x));
+ }
+
+ Assert.assertEquals(expected, actual);
+
+ periodicStorage.deletePeriodicQueryResults(id, 1L);
+
+ Set<BindingSet> actual2 = new HashSet<>();
+ try(CloseableIterator<BindingSet> iter = periodicStorage.listResults(id, Optional.of(1L))) {
+ iter.forEachRemaining(x -> actual2.add(x));
+ }
+
+ Assert.assertEquals(new HashSet<>(), actual2);
+ periodicStorage.deletePeriodicQuery(id);
+
+ }
+
+ @Test
+ public void multiBinTest() throws PeriodicQueryStorageException, Exception {
+
+ String sparql = "prefix function: <http://org.apache.rya/function#> " //n
+ + "prefix time: <http://www.w3.org/2006/time#> " //n
+ + "select ?id (count(?obs) as ?total) where {" //n
+ + "Filter(function:periodic(?time, 2, .5, time:hours)) " //n
+ + "?obs <uri:hasTime> ?time. " //n
+ + "?obs <uri:hasId> ?id } group by ?id"; //n
+
+
+ final ValueFactory vf = new ValueFactoryImpl();
+ long currentTime = System.currentTimeMillis();
+ String queryId = UUID.randomUUID().toString().replace("-", "");
+
+ // Create the expected results of the SPARQL query once the PCJ has been computed.
+ final Set<BindingSet> expected1 = new HashSet<>();
+ final Set<BindingSet> expected2 = new HashSet<>();
+ final Set<BindingSet> expected3 = new HashSet<>();
+ final Set<BindingSet> expected4 = new HashSet<>();
+ final Set<VisibilityBindingSet> storageResults = new HashSet<>();
+
+ long period = 1800000;
+ long binId = (currentTime/period)*period;
+
+ MapBindingSet bs = new MapBindingSet();
+ bs.addBinding("total", vf.createLiteral("2", XMLSchema.INTEGER));
+ bs.addBinding("id", vf.createLiteral("id_1", XMLSchema.STRING));
+ bs.addBinding("periodicBinId", vf.createLiteral(binId));
+ expected1.add(bs);
+ storageResults.add(new VisibilityBindingSet(bs));
+
+ bs = new MapBindingSet();
+ bs.addBinding("total", vf.createLiteral("2", XMLSchema.INTEGER));
+ bs.addBinding("id", vf.createLiteral("id_2", XMLSchema.STRING));
+ bs.addBinding("periodicBinId", vf.createLiteral(binId));
+ expected1.add(bs);
+ storageResults.add(new VisibilityBindingSet(bs));
+
+ bs = new MapBindingSet();
+ bs.addBinding("total", vf.createLiteral("1", XMLSchema.INTEGER));
+ bs.addBinding("id", vf.createLiteral("id_3", XMLSchema.STRING));
+ bs.addBinding("periodicBinId", vf.createLiteral(binId));
+ expected1.add(bs);
+ storageResults.add(new VisibilityBindingSet(bs));
+
+ bs = new MapBindingSet();
+ bs.addBinding("total", vf.createLiteral("1", XMLSchema.INTEGER));
+ bs.addBinding("id", vf.createLiteral("id_4", XMLSchema.STRING));
+ bs.addBinding("periodicBinId", vf.createLiteral(binId));
+ expected1.add(bs);
+ storageResults.add(new VisibilityBindingSet(bs));
+
+ bs = new MapBindingSet();
+ bs.addBinding("total", vf.createLiteral("1", XMLSchema.INTEGER));
+ bs.addBinding("id", vf.createLiteral("id_1", XMLSchema.STRING));
+ bs.addBinding("periodicBinId", vf.createLiteral(binId + period));
+ expected2.add(bs);
+ storageResults.add(new VisibilityBindingSet(bs));
+
+ bs = new MapBindingSet();
+ bs.addBinding("total", vf.createLiteral("2", XMLSchema.INTEGER));
+ bs.addBinding("id", vf.createLiteral("id_2", XMLSchema.STRING));
+ bs.addBinding("periodicBinId", vf.createLiteral(binId + period));
+ expected2.add(bs);
+ storageResults.add(new VisibilityBindingSet(bs));
+
+ bs = new MapBindingSet();
+ bs.addBinding("total", vf.createLiteral("1", XMLSchema.INTEGER));
+ bs.addBinding("id", vf.createLiteral("id_3", XMLSchema.STRING));
+ bs.addBinding("periodicBinId", vf.createLiteral(binId + period));
+ expected2.add(bs);
+ storageResults.add(new VisibilityBindingSet(bs));
+
+ bs = new MapBindingSet();
+ bs.addBinding("total", vf.createLiteral("1", XMLSchema.INTEGER));
+ bs.addBinding("id", vf.createLiteral("id_1", XMLSchema.STRING));
+ bs.addBinding("periodicBinId", vf.createLiteral(binId + 2*period));
+ expected3.add(bs);
+ storageResults.add(new VisibilityBindingSet(bs));
+
+ bs = new MapBindingSet();
+ bs.addBinding("total", vf.createLiteral("1", XMLSchema.INTEGER));
+ bs.addBinding("id", vf.createLiteral("id_2", XMLSchema.STRING));
+ bs.addBinding("periodicBinId", vf.createLiteral(binId + 2*period));
+ expected3.add(bs);
+ storageResults.add(new VisibilityBindingSet(bs));
+
+ bs = new MapBindingSet();
+ bs.addBinding("total", vf.createLiteral("1", XMLSchema.INTEGER));
+ bs.addBinding("id", vf.createLiteral("id_1", XMLSchema.STRING));
+ bs.addBinding("periodicBinId", vf.createLiteral(binId + 3*period));
+ expected4.add(bs);
+ storageResults.add(new VisibilityBindingSet(bs));
+
+
+ String id = periodicStorage.createPeriodicQuery(queryId, sparql);
+ periodicStorage.addPeriodicQueryResults(queryId, storageResults);
+
+ try(CloseableIterator<BindingSet> iter = periodicStorage.listResults(queryId, Optional.of(binId))) {
+ Set<BindingSet> actual1 = new HashSet<>();
+ while(iter.hasNext()) {
+ actual1.add(iter.next());
+ }
+ Assert.assertEquals(expected1, actual1);
+ }
+
+ periodicStorage.deletePeriodicQueryResults(queryId, binId);
+ try(CloseableIterator<BindingSet> iter = periodicStorage.listResults(queryId, Optional.of(binId))) {
+ Set<BindingSet> actual1 = new HashSet<>();
+ while(iter.hasNext()) {
+ actual1.add(iter.next());
+ }
+ Assert.assertEquals(Collections.emptySet(), actual1);
+ }
+
+ try(CloseableIterator<BindingSet> iter = periodicStorage.listResults(queryId, Optional.of(binId + period))) {
+ Set<BindingSet> actual2 = new HashSet<>();
+ while(iter.hasNext()) {
+ actual2.add(iter.next());
+ }
+ Assert.assertEquals(expected2, actual2);
+ }
+
+ periodicStorage.deletePeriodicQueryResults(queryId, binId + period);
+ try(CloseableIterator<BindingSet> iter = periodicStorage.listResults(queryId, Optional.of(binId + period))) {
+ Set<BindingSet> actual2 = new HashSet<>();
+ while(iter.hasNext()) {
+ actual2.add(iter.next());
+ }
+ Assert.assertEquals(Collections.emptySet(), actual2);
+ }
+
+ try(CloseableIterator<BindingSet> iter = periodicStorage.listResults(queryId, Optional.of(binId + 2*period))) {
+ Set<BindingSet> actual3 = new HashSet<>();
+ while(iter.hasNext()) {
+ actual3.add(iter.next());
+ }
+ Assert.assertEquals(expected3, actual3);
+ }
+
+ try(CloseableIterator<BindingSet> iter = periodicStorage.listResults(queryId, Optional.of(binId + 3*period))) {
+ Set<BindingSet> actual4 = new HashSet<>();
+ while(iter.hasNext()) {
+ actual4.add(iter.next());
+ }
+ Assert.assertEquals(expected4, actual4);
+ }
+ periodicStorage.deletePeriodicQuery(id);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/README.md
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/README.md b/extras/rya.pcj.fluo/README.md
index 70361c1..1207705 100644
--- a/extras/rya.pcj.fluo/README.md
+++ b/extras/rya.pcj.fluo/README.md
@@ -19,7 +19,15 @@ Rya Incrementally Updating Precomputed Joins
============================================
This project is an implementation of the Rya Precomputed Join (PCJ) indexing
feature that runs on top of [Fluo][1] so that it may incrementally update the
-results of a query as new semantic triples are added to storage.
+results of a query as new semantic triples are added to storage. At a high level, the Rya Fluo application
+works by registering the individual RDF4J QueryNodes with the Fluo table in the form of metadata. For example,
+if a join occurs in a given query, then that join is given a unique id when the query is registered with the Rya
+Fluo application, along with metadata indicating its parent node, its left and right child nodes, along with
+other information necessary for the application to process the join. In this way, the entire RDF4J query tree is recreated
+within Fluo. For each node type supported by the Rya Fluo application, there is also an associated Fluo Observer
+that processes BindingSet notifications for that node (this occurs when a new result percolates up the query tree and
+arrives at that node in the form of a BindingSet). These Observers incrementally evaluate the queries registered with the
+Fluo application by performing the processing required for their associated node as soon as a result for that node is available.
This project contains the following modules:
* **rya.pcj.fluo.app** - A Fluo application that incrementally updates the results
@@ -38,5 +46,20 @@ This project contains the following modules:
* **integration** - Contains integration tests that use a MiniAccumuloCluster
and MiniFluo to ensure the Rya PCJ Fluo App work within an emulation of the
production environment.
+
+
+Currently the Rya Fluo Application supports RDF4J queries that contain Joins, Filters, Projections, StatementPatterns, and Aggregations.
+To support the evaluation of additional RDF4J query nodes in the Fluo application, here are the steps that need to be followed:
+
+ 1. Create the appropriate Metadata Object by extending CommonNodeMetadata (e.g. StatementPatternMetadata, JoinMetadata, etc.)
+ 2. Add metadata Columns to FluoQueryColumns
+ 3. Create NodeType from the metadata Columns
+ 4. Add the node prefix to IncrementalUpdateConstants
+ 5. Integrate metadata with FluoQueryMetadataDAO
+ 6. Create Updater and integrate with BindingSetUpdater
+ 7. Create Observer (e.g. StatementPatternObserver, JoinObserver, etc.)
+ 8. Integrate with SparqlFluoQueryBuilder
+
+All of the classes mentioned above can be found in the rya.pcj.fluo.app project.
[1]: http://fluo.io/
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/pcj.fluo.api/src/main/java/org/apache/rya/indexing/pcj/fluo/api/CreatePcj.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.api/src/main/java/org/apache/rya/indexing/pcj/fluo/api/CreatePcj.java b/extras/rya.pcj.fluo/pcj.fluo.api/src/main/java/org/apache/rya/indexing/pcj/fluo/api/CreatePcj.java
index a17f02f..767d9d2 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.api/src/main/java/org/apache/rya/indexing/pcj/fluo/api/CreatePcj.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.api/src/main/java/org/apache/rya/indexing/pcj/fluo/api/CreatePcj.java
@@ -26,6 +26,7 @@ import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.HashSet;
import java.util.List;
+import java.util.Optional;
import java.util.Set;
import org.apache.accumulo.core.client.AccumuloException;
@@ -44,9 +45,11 @@ 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.IncrementalUpdateConstants;
import org.apache.rya.indexing.pcj.fluo.app.query.FluoQuery;
import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryColumns;
import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryMetadataDAO;
+import org.apache.rya.indexing.pcj.fluo.app.query.QueryMetadata;
import org.apache.rya.indexing.pcj.fluo.app.query.SparqlFluoQueryBuilder;
import org.apache.rya.indexing.pcj.fluo.app.query.SparqlFluoQueryBuilder.NodeIds;
import org.apache.rya.indexing.pcj.fluo.app.query.StatementPatternMetadata;
@@ -62,6 +65,8 @@ import org.openrdf.query.algebra.StatementPattern;
import org.openrdf.query.parser.ParsedQuery;
import org.openrdf.query.parser.sparql.SPARQLParser;
+import com.google.common.base.Preconditions;
+
import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
import edu.umd.cs.findbugs.annotations.NonNull;
@@ -152,13 +157,51 @@ public class CreatePcj {
+
/**
- * Tells the Fluo PCJ Updater application to maintain a new PCJ. This method requires that a
- * PCJ table already exist for the query corresponding to the pcjId. Results will be exported
- * to this table.
+ * 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 or
+ * some other external resource. The export id is equivalent to the queryId that is returned,
+ * which is in contrast to the other createPcj methods in this class which accept an external pcjId
+ * that is used to identify the Accumulo table or Kafka topic for exporting results.
+ *
+ * @param sparql - sparql query String to be registered with Fluo
+ * @param fluo - A connection to the Fluo application that updates the PCJ index. (not null)
+ * @return queryId - The id of the root of the query metadata tree in Fluo
+ * @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}.
+ */
+ public String createPcj(String sparql, FluoClient fluo) throws MalformedQueryException {
+ Preconditions.checkNotNull(sparql);
+ Preconditions.checkNotNull(fluo);
+
+ FluoQuery fluoQuery = makeFluoQuery(sparql);
+ String queryId = null;
+ if(fluoQuery.getQueryMetadata().isPresent()) {
+ queryId = fluoQuery.getQueryMetadata().get().getNodeId();
+ queryId = queryId.split(IncrementalUpdateConstants.QUERY_PREFIX)[1];
+ } else {
+ queryId = fluoQuery.getConstructQueryMetadata().get().getNodeId();
+ queryId = queryId.split(IncrementalUpdateConstants.CONSTRUCT_PREFIX)[1];
+ }
+
+ String[] idArray = queryId.split("_");
+ String id = idArray[idArray.length - 1];
+
+ writeFluoQuery(fluo, fluoQuery, id);
+ return id;
+ }
+
+ /**
+ * 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 and be incrementally generated
+ * inside of Fluo. This method assumes that the user will export the results to Kafka or
+ * some other external resource.
*
* @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 sparql - sparql query String to be registered with Fluo
* @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.
@@ -166,40 +209,113 @@ public class CreatePcj {
*/
public FluoQuery createPcj(
final String pcjId,
- final PrecomputedJoinStorage pcjStorage,
+ final String sparql,
final FluoClient fluo) throws MalformedQueryException, PcjException {
requireNonNull(pcjId);
- requireNonNull(pcjStorage);
+ requireNonNull(sparql);
requireNonNull(fluo);
+ FluoQuery fluoQuery = makeFluoQuery(sparql);
+ writeFluoQuery(fluo, fluoQuery, pcjId);
+
+ return fluoQuery;
+ }
+
+ private FluoQuery makeFluoQuery(String sparql) throws MalformedQueryException {
+
// Keeps track of the IDs that are assigned to each of the query's nodes in Fluo.
// We use these IDs later when scanning Rya for historic Statement Pattern matches
// as well as setting up automatic exports.
final NodeIds nodeIds = new NodeIds();
// 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();
final ParsedQuery parsedQuery = new SPARQLParser().parseQuery(sparql, null);
- final FluoQuery fluoQuery = new SparqlFluoQueryBuilder().make(parsedQuery, nodeIds);
-
+ return new SparqlFluoQueryBuilder().make(parsedQuery, nodeIds);
+ }
+
+ 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);
-
- if (fluoQuery.getQueryMetadata().isPresent()) {
- // If the query is not a construct query,
- // the results of the query are eventually exported to an instance of Rya, so store the Rya ID for the PCJ.
- final String queryId = fluoQuery.getQueryMetadata().get().getNodeId();
+
+ // The results of the query are eventually exported to an instance
+ // of Rya, so store the Rya ID for the PCJ.
+ QueryMetadata metadata = fluoQuery.getQueryMetadata().orNull();
+ if (metadata != null) {
+ String queryId = metadata.getNodeId();
tx.set(queryId, FluoQueryColumns.RYA_PCJ_ID, pcjId);
tx.set(pcjId, FluoQueryColumns.PCJ_ID_QUERY_ID, queryId);
- }
+ }
+
// Flush the changes to Fluo.
tx.commit();
}
+ }
- 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.
+ *
+ * @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}.
+ */
+ public FluoQuery createPcj(
+ final String pcjId,
+ final PrecomputedJoinStorage pcjStorage,
+ final FluoClient fluo) throws MalformedQueryException, PcjException {
+ 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, fluo);
+ }
+
+ /**
+ * 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. It is assumed that results for any query registered
+ * using this method will be exported to Kafka or some other external service.
+ *
+ * @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 fluo - A connection to the Fluo application that updates the PCJ index. (not null)
+ * @param queryEngine - QueryEngine for a given Rya Instance, (not null)
+ * @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}.
+ */
+ public String withRyaIntegration(
+ final String pcjId,
+ final String sparql,
+ final FluoClient fluo,
+ final Connector accumulo,
+ final String ryaInstance ) throws MalformedQueryException, PcjException, RyaDAOException {
+ 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, 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().get().getNodeId();
}
+
/**
* Tells the Fluo PCJ Updater application to maintain a new PCJ.
@@ -231,31 +347,39 @@ public class CreatePcj {
requireNonNull(fluo);
requireNonNull(accumulo);
requireNonNull(ryaInstance);
-
- // Write the SPARQL query's structure to the Fluo Application.
- final FluoQuery fluoQuery = createPcj(pcjId, pcjStorage, 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 withRyaIntegration(pcjId, sparql, 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.
+ // 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.
+ // 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
+ // 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);
+ 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) {
+ for (final RyaStatement ryaStatement : queryIterable) {
if (triplesBatch.size() == spInsertBatchSize) {
writeBatch(fluo, triplesBatch);
triplesBatch.clear();
@@ -271,14 +395,6 @@ public class CreatePcj {
} catch (final IOException e) {
log.warn("Ignoring IOException thrown while closing the AccumuloRyaQueryEngine used by CreatePCJ.", e);
}
-
- //return queryId to the caller for later monitoring from the export
- if(fluoQuery.getConstructQueryMetadata().isPresent()) {
- return fluoQuery.getConstructQueryMetadata().get().getNodeId();
- }
-
- return fluoQuery.getQueryMetadata().get().getNodeId();
-
}
private static void writeBatch(final FluoClient fluo, final Set<RyaStatement> batch) {
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/pcj.fluo.api/src/main/java/org/apache/rya/indexing/pcj/fluo/api/DeletePcj.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.api/src/main/java/org/apache/rya/indexing/pcj/fluo/api/DeletePcj.java b/extras/rya.pcj.fluo/pcj.fluo.api/src/main/java/org/apache/rya/indexing/pcj/fluo/api/DeletePcj.java
index 87eb9cc..3052c1d 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.api/src/main/java/org/apache/rya/indexing/pcj/fluo/api/DeletePcj.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.api/src/main/java/org/apache/rya/indexing/pcj/fluo/api/DeletePcj.java
@@ -39,6 +39,7 @@ import org.apache.rya.indexing.pcj.fluo.app.query.FilterMetadata;
import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryColumns;
import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryMetadataDAO;
import org.apache.rya.indexing.pcj.fluo.app.query.JoinMetadata;
+import org.apache.rya.indexing.pcj.fluo.app.query.PeriodicQueryMetadata;
import org.apache.rya.indexing.pcj.fluo.app.query.QueryMetadata;
import org.openrdf.query.BindingSet;
@@ -50,12 +51,12 @@ import edu.umd.cs.findbugs.annotations.NonNull;
* <p>
* This is a two phase process.
* <ol>
- * <li>Delete metadata about each node of the query using a single Fluo
- * transaction. This prevents new {@link BindingSet}s from being created when
- * new triples are inserted.</li>
- * <li>Delete BindingSets associated with each node of the query. This is done
- * in a batch fashion to guard against large delete transactions that don't fit
- * into memory.</li>
+ * <li>Delete metadata about each node of the query using a single Fluo
+ * transaction. This prevents new {@link BindingSet}s from being created when
+ * new triples are inserted.</li>
+ * <li>Delete BindingSets associated with each node of the query. This is done
+ * in a batch fashion to guard against large delete transactions that don't fit
+ * into memory.</li>
* </ol>
*/
@DefaultAnnotation(NonNull.class)
@@ -79,8 +80,10 @@ public class DeletePcj {
* Precomputed Join Index from the Fluo application that is incrementally
* updating it.
*
- * @param client - Connects to the Fluo application that is updating the PCJ Index. (not null)
- * @param pcjId - The PCJ ID for the query that will removed from the Fluo application. (not null)
+ * @param client - Connects to the Fluo application that is updating the PCJ
+ * Index. (not null)
+ * @param pcjId - The PCJ ID for the query that will removed from the Fluo
+ * application. (not null)
*/
public void deletePcj(final FluoClient client, final String pcjId) {
requireNonNull(client);
@@ -167,6 +170,12 @@ public class DeletePcj {
nodeIds.add(aggChild);
getChildNodeIds(tx, aggChild, nodeIds);
break;
+ case PERIODIC_QUERY:
+ final PeriodicQueryMetadata periodicMeta = dao.readPeriodicQueryMetadata(tx, nodeId);
+ final String periodicChild = periodicMeta.getChildNodeId();
+ nodeIds.add(periodicChild);
+ getChildNodeIds(tx, periodicChild, nodeIds);
+ break;
case STATEMENT_PATTERN:
break;
}
@@ -215,10 +224,9 @@ public class DeletePcj {
}
}
-
/**
- * Deletes high level query meta for converting from queryId to pcjId and vice
- * versa, as well as converting from sparql to queryId.
+ * Deletes high level query meta for converting from queryId to pcjId and
+ * vice versa, as well as converting from sparql to queryId.
*
* @param tx - Transaction the deletes will be performed with. (not null)
* @param pcjId - The PCJ whose metadata will be deleted. (not null)
@@ -234,7 +242,6 @@ public class DeletePcj {
tx.delete(pcjId, FluoQueryColumns.PCJ_ID_QUERY_ID);
}
-
/**
* Deletes all results (BindingSets or Statements) associated with the specified nodeId.
*
@@ -265,18 +272,18 @@ public class DeletePcj {
requireNonNull(scanner);
requireNonNull(column);
- try(Transaction ntx = tx) {
- int count = 0;
- final Iterator<RowColumnValue> iter = scanner.iterator();
- while (iter.hasNext() && count < batchSize) {
- final Bytes row = iter.next().getRow();
- count++;
- tx.delete(row, column);
- }
+ try (Transaction ntx = tx) {
+ int count = 0;
+ final Iterator<RowColumnValue> iter = scanner.iterator();
+ while (iter.hasNext() && count < batchSize) {
+ final Bytes row = iter.next().getRow();
+ count++;
+ tx.delete(row, column);
+ }
- final boolean hasNext = iter.hasNext();
- tx.commit();
- return hasNext;
+ final boolean hasNext = iter.hasNext();
+ tx.commit();
+ return hasNext;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/pcj.fluo.app/pom.xml
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/pom.xml b/extras/rya.pcj.fluo/pcj.fluo.app/pom.xml
index 38fff95..b151c0e 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/pom.xml
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/pom.xml
@@ -1,35 +1,28 @@
<?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
+<!-- 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">
- http://www.apache.org/licenses/LICENSE-2.0
+ <parent>
+ <groupId>org.apache.rya</groupId>
+ <artifactId>rya.pcj.fluo.parent</artifactId>
+ <version>3.2.11-incubating-SNAPSHOT</version>
+ </parent>
-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">
+ <modelVersion>4.0.0</modelVersion>
+ <artifactId>rya.pcj.fluo.app</artifactId>
- <parent>
- <groupId>org.apache.rya</groupId>
- <artifactId>rya.pcj.fluo.parent</artifactId>
- <version>3.2.11-incubating-SNAPSHOT</version>
- </parent>
-
- <modelVersion>4.0.0</modelVersion>
- <artifactId>rya.pcj.fluo.app</artifactId>
-
- <name>Apache Rya PCJ Fluo App</name>
- <description>
+ <name>Apache Rya PCJ Fluo App</name>
+ <description>
A Fluo implementation of Rya Precomputed Join Indexing. This module produces
a jar that may be executed by the 'fluo' command line tool as a YARN job.
</description>
@@ -72,6 +65,10 @@ under the License.
<groupId>org.apache.fluo</groupId>
<artifactId>fluo-recipes-accumulo</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.openrdf.sesame</groupId>
+ <artifactId>sesame-queryrender</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
@@ -123,27 +120,29 @@ under the License.
</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>
- <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>
+
+ <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>
+ <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>
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/FilterFinder.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/FilterFinder.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/FilterFinder.java
deleted file mode 100644
index ae976ee..0000000
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/FilterFinder.java
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
- * 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.app;
-
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
-
-import java.util.concurrent.atomic.AtomicReference;
-
-import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
-import edu.umd.cs.findbugs.annotations.NonNull;
-
-import org.openrdf.query.algebra.Filter;
-import org.openrdf.query.algebra.helpers.QueryModelVisitorBase;
-import org.openrdf.query.parser.ParsedQuery;
-import org.openrdf.query.parser.sparql.SPARQLParser;
-
-import com.google.common.base.Optional;
-
-/**
- * Searches a SPARQL query for {@link Filter}s.
- */
-@DefaultAnnotation(NonNull.class)
-class FilterFinder {
-
- /**
- * Search a SPARQL query for the {@link Filter} that appears at the
- * {@code indexWithinQuery}'th time within the query.
- * <p>
- * The top most filter within the query will be at index 0, the next filter
- * encountered will be at index 1, ... and the last index that is encountered
- * will be at index <i>n</i>.
- *
- * @param sparql - The SPARQL query that to parse. (not null)
- * @param indexWithinQuery - The index of the filter to fetch. (not null)
- * @return The filter that was found within the query at the specified index;
- * otherwise absent.
- * @throws Exception Thrown when the query could not be parsed or iterated over.
- */
- public Optional<Filter> findFilter(final String sparql, final int indexWithinQuery) throws Exception {
- checkNotNull(sparql);
- checkArgument(indexWithinQuery >= 0);
-
- // When a filter is encountered for the requested index, store it in atomic reference and quit searching.
- final AtomicReference<Filter> filterRef = new AtomicReference<>();
- final QueryModelVisitorBase<RuntimeException> filterFinder = new QueryModelVisitorBase<RuntimeException>() {
- private int i = 0;
- @Override
- public void meet(final Filter filter) {
- // Store and stop searching.
- if(i == indexWithinQuery) {
- filterRef.set(filter);
- return;
- }
-
- // Continue to the next filter.
- i++;
- super.meet(filter);
- }
- };
-
- // Parse the query and find the filter.
- final SPARQLParser parser = new SPARQLParser();
- final ParsedQuery parsedQuery = parser.parseQuery(sparql, null);
- parsedQuery.getTupleExpr().visit(filterFinder);
- return Optional.fromNullable(filterRef.get());
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/FilterResultUpdater.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/FilterResultUpdater.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/FilterResultUpdater.java
index 42ec686..1c99051 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/FilterResultUpdater.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/FilterResultUpdater.java
@@ -26,9 +26,11 @@ import org.apache.log4j.Logger;
import org.apache.rya.indexing.pcj.fluo.app.query.FilterMetadata;
import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryColumns;
import org.apache.rya.indexing.pcj.fluo.app.util.BindingSetUtil;
+import org.apache.rya.indexing.pcj.fluo.app.util.FilterSerializer;
import org.apache.rya.indexing.pcj.fluo.app.util.RowKeyUtil;
import org.apache.rya.indexing.pcj.storage.accumulo.VariableOrder;
import org.apache.rya.indexing.pcj.storage.accumulo.VisibilityBindingSet;
+import org.apache.rya.indexing.pcj.storage.accumulo.VisibilityBindingSetSerDe;
import org.openrdf.model.Resource;
import org.openrdf.model.Statement;
import org.openrdf.model.URI;
@@ -62,11 +64,6 @@ public class FilterResultUpdater {
private static final VisibilityBindingSetSerDe BS_SERDE = new VisibilityBindingSetSerDe();
/**
- * A utility class used to search SPARQL queries for Filters.
- */
- private static final FilterFinder filterFinder = new FilterFinder();
-
- /**
* Is used to evaluate the conditions of a {@link Filter}.
*/
private static final EvaluationStrategyImpl evaluator = new EvaluationStrategyImpl(
@@ -111,12 +108,11 @@ public class FilterResultUpdater {
"Binding Set:\n" + childBindingSet + "\n");
// Parse the original query and find the Filter that represents filterId.
- final String sparql = filterMetadata.getOriginalSparql();
- final int indexWithinQuery = filterMetadata.getFilterIndexWithinSparql();
- final Optional<Filter> filter = filterFinder.findFilter(sparql, indexWithinQuery);
+ final String sparql = filterMetadata.getFilterSparql();
+ Filter filter = FilterSerializer.deserialize(sparql);
// Evaluate whether the child BindingSet satisfies the filter's condition.
- final ValueExpr condition = filter.get().getCondition();
+ final ValueExpr condition = filter.getCondition();
if (isTrue(condition, childBindingSet)) {
// Create the Filter's binding set from the child's.
final VariableOrder filterVarOrder = filterMetadata.getVariableOrder();
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/IncrementalUpdateConstants.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/IncrementalUpdateConstants.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/IncrementalUpdateConstants.java
index f9d14b5..2084907 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/IncrementalUpdateConstants.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/IncrementalUpdateConstants.java
@@ -18,6 +18,8 @@
*/
package org.apache.rya.indexing.pcj.fluo.app;
+import org.apache.rya.indexing.pcj.storage.PeriodicQueryResultStorage;
+
public class IncrementalUpdateConstants {
// String constants used to create more easily parsed patterns.
@@ -34,6 +36,9 @@ public class IncrementalUpdateConstants {
public static final String AGGREGATION_PREFIX = "AGGREGATION";
public static final String QUERY_PREFIX = "QUERY";
public static final String CONSTRUCT_PREFIX = "CONSTRUCT";
+ public static final String PERIODIC_QUERY_PREFIX = "PERIODIC_QUERY";
+
+ public static final String PERIODIC_BIN_ID = PeriodicQueryResultStorage.PeriodicBinId;
public static final String URI_TYPE = "http://www.w3.org/2001/XMLSchema#anyURI";
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/JoinResultUpdater.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/JoinResultUpdater.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/JoinResultUpdater.java
index 2cb5a54..9b65b34 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/JoinResultUpdater.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/JoinResultUpdater.java
@@ -43,6 +43,7 @@ import org.apache.rya.indexing.pcj.fluo.app.util.RowKeyUtil;
import org.apache.rya.indexing.pcj.storage.accumulo.BindingSetConverter.BindingSetConversionException;
import org.apache.rya.indexing.pcj.storage.accumulo.VariableOrder;
import org.apache.rya.indexing.pcj.storage.accumulo.VisibilityBindingSet;
+import org.apache.rya.indexing.pcj.storage.accumulo.VisibilityBindingSetSerDe;
import org.apache.rya.indexing.pcj.storage.accumulo.VisibilityBindingSetStringConverter;
import org.openrdf.query.Binding;
import org.openrdf.query.BindingSet;
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/NodeType.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/NodeType.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/NodeType.java
index b829b7e..b8fc2d9 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/NodeType.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/NodeType.java
@@ -25,6 +25,7 @@ import static org.apache.rya.indexing.pcj.fluo.app.IncrementalUpdateConstants.FI
import static org.apache.rya.indexing.pcj.fluo.app.IncrementalUpdateConstants.JOIN_PREFIX;
import static org.apache.rya.indexing.pcj.fluo.app.IncrementalUpdateConstants.QUERY_PREFIX;
import static org.apache.rya.indexing.pcj.fluo.app.IncrementalUpdateConstants.SP_PREFIX;
+import static org.apache.rya.indexing.pcj.fluo.app.IncrementalUpdateConstants.PERIODIC_QUERY_PREFIX;
import java.util.List;
@@ -38,6 +39,7 @@ import com.google.common.base.Optional;
* Represents the different types of nodes that a Query may have.
*/
public enum NodeType {
+ PERIODIC_QUERY(QueryNodeMetadataColumns.PERIODIC_QUERY_COLUMNS, FluoQueryColumns.PERIODIC_QUERY_BINDING_SET),
FILTER (QueryNodeMetadataColumns.FILTER_COLUMNS, FluoQueryColumns.FILTER_BINDING_SET),
JOIN(QueryNodeMetadataColumns.JOIN_COLUMNS, FluoQueryColumns.JOIN_BINDING_SET),
STATEMENT_PATTERN(QueryNodeMetadataColumns.STATEMENTPATTERN_COLUMNS, FluoQueryColumns.STATEMENT_PATTERN_BINDING_SET),
@@ -101,6 +103,8 @@ public enum NodeType {
type = AGGREGATION;
} else if(nodeId.startsWith(CONSTRUCT_PREFIX)) {
type = CONSTRUCT;
+ } else if(nodeId.startsWith(PERIODIC_QUERY_PREFIX)) {
+ type = PERIODIC_QUERY;
}
return Optional.fromNullable(type);
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/PeriodicQueryUpdater.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/PeriodicQueryUpdater.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/PeriodicQueryUpdater.java
new file mode 100644
index 0000000..ae4912b
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/PeriodicQueryUpdater.java
@@ -0,0 +1,138 @@
+/*
+ * 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.app;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.fluo.api.client.TransactionBase;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+import org.apache.log4j.Logger;
+import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryColumns;
+import org.apache.rya.indexing.pcj.fluo.app.query.PeriodicQueryMetadata;
+import org.apache.rya.indexing.pcj.fluo.app.util.RowKeyUtil;
+import org.apache.rya.indexing.pcj.storage.accumulo.VisibilityBindingSet;
+import org.apache.rya.indexing.pcj.storage.accumulo.VisibilityBindingSetSerDe;
+import org.openrdf.model.Literal;
+import org.openrdf.model.Value;
+import org.openrdf.model.ValueFactory;
+import org.openrdf.model.impl.ValueFactoryImpl;
+import org.openrdf.query.Binding;
+import org.openrdf.query.algebra.evaluation.QueryBindingSet;
+
+/**
+ * This class adds the appropriate BinId Binding to each BindingSet that it processes. The BinIds
+ * are used to determine which period a BindingSet (with a temporal Binding) falls into so that
+ * a user can receive periodic updates for a registered query.
+ *
+ */
+public class PeriodicQueryUpdater {
+
+ private static final Logger log = Logger.getLogger(PeriodicQueryUpdater.class);
+ private static final ValueFactory vf = new ValueFactoryImpl();
+ private static final VisibilityBindingSetSerDe BS_SERDE = new VisibilityBindingSetSerDe();
+
+ /**
+ * Uses the {@link PeriodicQueryMetadata} to create a collection of binned BindingSets
+ * that are added to Fluo. Each binned BindingSet is the original BindingSet with an additional
+ * Binding that contains the periodic bin id of the BindingSet.
+ * @param tx - Fluo Transaction
+ * @param bs - VisibilityBindingSet that will be binned
+ * @param metadata - PeriodicQueryMetadata used to bin BindingSets
+ * @throws Exception
+ */
+ public void updatePeriodicBinResults(TransactionBase tx, VisibilityBindingSet bs, PeriodicQueryMetadata metadata) throws Exception {
+ Set<Long> binIds = getBinEndTimes(metadata, bs);
+ for(Long id: binIds) {
+ //create binding set value bytes
+ QueryBindingSet binnedBs = new QueryBindingSet(bs);
+ binnedBs.addBinding(IncrementalUpdateConstants.PERIODIC_BIN_ID, vf.createLiteral(id));
+ VisibilityBindingSet visibilityBindingSet = new VisibilityBindingSet(binnedBs, bs.getVisibility());
+ Bytes periodicBsBytes = BS_SERDE.serialize(visibilityBindingSet);
+
+ //create row
+ final Bytes resultRow = RowKeyUtil.makeRowKey(metadata.getNodeId(), metadata.getVariableOrder(), binnedBs);
+ Column col = FluoQueryColumns.PERIODIC_QUERY_BINDING_SET;
+ tx.set(resultRow, col, periodicBsBytes);
+ }
+ }
+
+ /**
+ * This method returns the end times of all period windows containing the time contained in
+ * the BindingSet.
+ *
+ * @param metadata
+ * @return Set of period bin end times
+ */
+ private Set<Long> getBinEndTimes(PeriodicQueryMetadata metadata, VisibilityBindingSet bs) {
+ Set<Long> binIds = new HashSet<>();
+ try {
+ String timeVar = metadata.getTemporalVariable();
+ Value value = bs.getBinding(timeVar).getValue();
+ Literal temporalLiteral = (Literal) value;
+ long eventDateTime = temporalLiteral.calendarValue().toGregorianCalendar().getTimeInMillis();
+ return getEndTimes(eventDateTime, metadata.getWindowSize(), metadata.getPeriod());
+ } catch (Exception e) {
+ log.trace("Unable to extract the entity time from BindingSet: " + bs);
+ }
+ return binIds;
+ }
+
+ private long getRightBinEndPoint(long eventDateTime, long periodDuration) {
+ return (eventDateTime / periodDuration + 1) * periodDuration;
+ }
+
+ private long getLeftBinEndPoint(long eventTime, long periodDuration) {
+ return (eventTime / periodDuration) * periodDuration;
+ }
+
+ /**
+ * Using the smallest period end time, this method also creates all other period end times
+ * that occur within one windowSize of the eventDateTime.
+ * @param eventDateTime
+ * @param startTime
+ * @param windowDuration
+ * @param periodDuration
+ * @return Set of period bin end times
+ */
+ private Set<Long> getEndTimes(long eventDateTime, long windowDuration, long periodDuration) {
+ Set<Long> binIds = new HashSet<>();
+ long rightEventBin = getRightBinEndPoint(eventDateTime, periodDuration);
+ //get the bin left of the current moment for comparison
+ long currentBin = getLeftBinEndPoint(System.currentTimeMillis(), periodDuration);
+
+ if(currentBin >= rightEventBin) {
+ long numBins = (windowDuration -(currentBin - rightEventBin))/periodDuration;
+ for(int i = 0; i < numBins; i++) {
+ binIds.add(currentBin + i*periodDuration);
+ }
+ } else {
+ //this corresponds to a future event that is inserted into the system
+ long numBins = windowDuration/periodDuration;
+ for(int i = 0; i < numBins; i++) {
+ binIds.add(rightEventBin + i*periodDuration);
+ }
+ }
+
+ return binIds;
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/QueryResultUpdater.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/QueryResultUpdater.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/QueryResultUpdater.java
index ba82726..44fc9bd 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/QueryResultUpdater.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/QueryResultUpdater.java
@@ -31,6 +31,7 @@ import org.apache.rya.indexing.pcj.fluo.app.util.BindingSetUtil;
import org.apache.rya.indexing.pcj.fluo.app.util.RowKeyUtil;
import org.apache.rya.indexing.pcj.storage.accumulo.VariableOrder;
import org.apache.rya.indexing.pcj.storage.accumulo.VisibilityBindingSet;
+import org.apache.rya.indexing.pcj.storage.accumulo.VisibilityBindingSetSerDe;
import org.openrdf.query.BindingSet;
import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
@@ -88,7 +89,7 @@ public class QueryResultUpdater {
}
// Create the Binding Set that goes in the Node Value. It does contain visibilities.
- final Bytes nodeValueBytes = BS_SERDE.serialize(childBindingSet);
+ final Bytes nodeValueBytes = BS_SERDE.serialize(new VisibilityBindingSet(queryBindingSet,childBindingSet.getVisibility()));
log.trace(
"Transaction ID: " + tx.getStartTimestamp() + "\n" +
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/VisibilityBindingSetSerDe.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/VisibilityBindingSetSerDe.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/VisibilityBindingSetSerDe.java
deleted file mode 100644
index 34439e4..0000000
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/VisibilityBindingSetSerDe.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * 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.app;
-
-import static java.util.Objects.requireNonNull;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
-
-import org.apache.fluo.api.data.Bytes;
-import org.apache.rya.indexing.pcj.storage.accumulo.VisibilityBindingSet;
-
-import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
-import edu.umd.cs.findbugs.annotations.NonNull;
-
-/**
- * Serializes and deserializes a {@link VisibilityBindingSet} to and from {@link Bytes} objects.
- */
-@DefaultAnnotation(NonNull.class)
-public class VisibilityBindingSetSerDe {
-
- /**
- * Serializes a {@link VisibilityBindingSet} into a {@link Bytes} object.
- *
- * @param bindingSet - The binding set that will be serialized. (not null)
- * @return The serialized object.
- * @throws Exception A problem was encountered while serializing the object.
- */
- public Bytes serialize(final VisibilityBindingSet bindingSet) throws Exception {
- requireNonNull(bindingSet);
-
- final ByteArrayOutputStream boas = new ByteArrayOutputStream();
- try(final ObjectOutputStream oos = new ObjectOutputStream(boas)) {
- oos.writeObject(bindingSet);
- }
-
- return Bytes.of(boas.toByteArray());
- }
-
- /**
- * Deserializes a {@link VisibilityBindingSet} from a {@link Bytes} object.
- *
- * @param bytes - The bytes that will be deserialized. (not null)
- * @return The deserialized object.
- * @throws Exception A problem was encountered while deserializing the object.
- */
- public VisibilityBindingSet deserialize(final Bytes bytes) throws Exception {
- requireNonNull(bytes);
-
- try(final ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes.toArray()))) {
- final Object o = ois.readObject();
- if(o instanceof VisibilityBindingSet) {
- return (VisibilityBindingSet) o;
- } else {
- throw new Exception("Deserialized Object is not a VisibilityBindingSet. Was: " + o.getClass());
- }
- }
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/AbstractBatchBindingSetUpdater.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/AbstractBatchBindingSetUpdater.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/AbstractBatchBindingSetUpdater.java
new file mode 100644
index 0000000..db33d3b
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/AbstractBatchBindingSetUpdater.java
@@ -0,0 +1,55 @@
+/*
+ * 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.app.batch;
+
+import org.apache.fluo.api.client.TransactionBase;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.RowColumn;
+import org.apache.fluo.api.data.Span;
+import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryColumns;
+
+/**
+ * This class provides common functionality for implementations of {@link BatchBindingSetUpdater}.
+ *
+ */
+public abstract class AbstractBatchBindingSetUpdater implements BatchBindingSetUpdater {
+
+ /**
+ * Updates the Span to create a new {@link BatchInformation} object to be fed to the
+ * {@link BatchObserver}. This message is called in the event that the BatchBindingSetUpdater
+ * reaches the batch size before processing all entries relevant to its Span.
+ * @param newStart - new start to the Span
+ * @param oldSpan - old Span to be updated with newStart
+ * @return - updated Span used with an updated BatchInformation object to complete the batch task
+ */
+ public Span getNewSpan(RowColumn newStart, Span oldSpan) {
+ return new Span(newStart, oldSpan.isStartInclusive(), oldSpan.getEnd(), oldSpan.isEndInclusive());
+ }
+
+ /**
+ * Cleans up old batch job. This method is meant to be called by any overriding method
+ * to clean up old batch tasks.
+ */
+ @Override
+ public void processBatch(TransactionBase tx, Bytes row, BatchInformation batch) throws Exception {
+ tx.delete(row, FluoQueryColumns.BATCH_COLUMN);
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/AbstractSpanBatchInformation.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/AbstractSpanBatchInformation.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/AbstractSpanBatchInformation.java
new file mode 100644
index 0000000..498dd85
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/AbstractSpanBatchInformation.java
@@ -0,0 +1,101 @@
+/*
+ * 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.app.batch;
+
+import java.util.Objects;
+
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.Span;
+
+import jline.internal.Preconditions;
+
+/**
+ * Abstract class for generating span based notifications. A spanned notification
+ * uses a {@link Span} to begin processing a Fluo Column at the position designated by the Span.
+ *
+ */
+public abstract class AbstractSpanBatchInformation extends BasicBatchInformation {
+
+ private Span span;
+
+ /**
+ * Create AbstractBatchInformation
+ * @param batchSize - size of batch to be processed
+ * @param task - type of task processed (Add, Delete, Udpate)
+ * @param column - Cpolumn that Span notification is applied
+ * @param span - span used to indicate where processing should begin
+ */
+ public AbstractSpanBatchInformation(int batchSize, Task task, Column column, Span span) {
+ super(batchSize, task, column);
+ this.span = Preconditions.checkNotNull(span);
+ }
+
+ public AbstractSpanBatchInformation(Task task, Column column, Span span) {
+ this(DEFAULT_BATCH_SIZE, task, column, span);
+ }
+
+ /**
+ * @return Span that batch Task will be applied to
+ */
+ public Span getSpan() {
+ return span;
+ }
+
+ /**
+ * Sets span to which batch Task will be applied
+ * @param span
+ */
+ public void setSpan(Span span) {
+ this.span = span;
+ }
+
+ @Override
+ public String toString() {
+ return new StringBuilder()
+ .append("Span Batch Information {\n")
+ .append(" Span: " + span + "\n")
+ .append(" Batch Size: " + super.getBatchSize() + "\n")
+ .append(" Task: " + super.getTask() + "\n")
+ .append(" Column: " + super.getColumn() + "\n")
+ .append("}")
+ .toString();
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (this == other) {
+ return true;
+ }
+
+ if (!(other instanceof AbstractSpanBatchInformation)) {
+ return false;
+ }
+
+ AbstractSpanBatchInformation batch = (AbstractSpanBatchInformation) other;
+ return (super.getBatchSize() == batch.getBatchSize()) && Objects.equals(super.getColumn(), batch.getColumn()) && Objects.equals(this.span, batch.span)
+ && Objects.equals(super.getTask(), batch.getTask());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.getBatchSize(), span, super.getColumn(), super.getTask());
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BasicBatchInformation.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BasicBatchInformation.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BasicBatchInformation.java
new file mode 100644
index 0000000..288ed6e
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BasicBatchInformation.java
@@ -0,0 +1,81 @@
+package org.apache.rya.indexing.pcj.fluo.app.batch;
+/*
+ * 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.
+ */
+import org.apache.fluo.api.data.Column;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * This class contains all of the common info contained in other implementations
+ * of BatchInformation.
+ *
+ */
+public abstract class BasicBatchInformation implements BatchInformation {
+
+ private int batchSize;
+ private Task task;
+ private Column column;
+
+ /**
+ * Create BasicBatchInformation object
+ * @param batchSize - size of batch to be processed
+ * @param task - task to be processed
+ * @param column - Column in which data is proessed
+ */
+ public BasicBatchInformation(int batchSize, Task task, Column column ) {
+ this.task = Preconditions.checkNotNull(task);
+ this.column = Preconditions.checkNotNull(column);
+ Preconditions.checkArgument(batchSize > 0);
+ this.batchSize = batchSize;
+ }
+
+ /**
+ * Creates a BasicBatchInformation
+ * @param task
+ */
+ public BasicBatchInformation(Task task, Column column) {
+ Preconditions.checkNotNull(task);
+ Preconditions.checkNotNull(column);
+ this.task = task;
+ this.column = column;
+ this.batchSize = DEFAULT_BATCH_SIZE;
+ }
+
+ /**
+ * @return - size of batch that tasks are performed in
+ */
+ public int getBatchSize() {
+ return batchSize;
+ }
+
+ /**
+ * @return - type of Task performed (Add, Delete, Update)
+ */
+ public Task getTask() {
+ return task;
+ }
+
+ /**
+ * @return - Column in which Task will be performed
+ */
+ public Column getColumn() {
+ return column;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BatchBindingSetUpdater.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BatchBindingSetUpdater.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BatchBindingSetUpdater.java
new file mode 100644
index 0000000..2076d2d
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BatchBindingSetUpdater.java
@@ -0,0 +1,43 @@
+package org.apache.rya.indexing.pcj.fluo.app.batch;
+/*
+ * 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.
+ */
+import org.apache.fluo.api.client.TransactionBase;
+import org.apache.fluo.api.data.Bytes;
+
+/**
+ * Interface for applying batch updates to the Fluo table based on the provided {@link BatchInformation}.
+ * This updater is used by the {@link BatchObserver} to apply batch updates to overcome the restriction
+ * that all transactions are processed in memory. This allows Observers process potentially large
+ * tasks that cannot fit into memory in a piece-wise, batch fashion.
+ */
+public interface BatchBindingSetUpdater {
+
+ /**
+ * Processes the {@link BatchInformation} object. The BatchInformation will
+ * typically include a Task (either Add, Update, or Delete), along with information
+ * about the starting point to begin processing data.
+ * @param tx - Fluo Transaction
+ * @param row - contains the ID of the Fluo node to be processed
+ * @param batch - contains info about which cells for the Fluo query result node
+ * should be processed
+ * @throws Exception
+ */
+ public void processBatch(TransactionBase tx, Bytes row, BatchInformation batch) throws Exception;
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BatchInformation.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BatchInformation.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BatchInformation.java
new file mode 100644
index 0000000..7b23ee7
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BatchInformation.java
@@ -0,0 +1,57 @@
+package org.apache.rya.indexing.pcj.fluo.app.batch;
+/*
+ * 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.
+ */
+import org.apache.fluo.api.data.Column;
+
+/**
+ * Interface for submitting batch Fluo tasks to be processed by the
+ * {@link BatchObserver}. The BatchObserver applies batch updates to overcome
+ * the restriction that all Fluo transactions are processed in memory. This
+ * allows the Rya Fluo application to process large tasks that cannot fit into
+ * memory in a piece-wise, batch fashion.
+ */
+public interface BatchInformation {
+
+ public static enum Task {Add, Delete, Update}
+ public static int DEFAULT_BATCH_SIZE = 5000;
+
+ /**
+ * @return batchsize of task
+ */
+ public int getBatchSize();
+
+ /**
+ *
+ * @return Task to be performed
+ */
+ public Task getTask();
+
+ /**
+ *
+ * @return Column that task will be performed on
+ */
+ public Column getColumn();
+
+ /**
+ *
+ * @return BatchBindingSetUpdater used to process this Batch Task
+ */
+ public BatchBindingSetUpdater getBatchUpdater();
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BatchInformationDAO.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BatchInformationDAO.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BatchInformationDAO.java
new file mode 100644
index 0000000..f9ed658
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BatchInformationDAO.java
@@ -0,0 +1,59 @@
+package org.apache.rya.indexing.pcj.fluo.app.batch;
+/*
+ * 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.
+ */
+import java.util.Optional;
+
+import org.apache.fluo.api.client.TransactionBase;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.rya.indexing.pcj.fluo.app.batch.serializer.BatchInformationSerializer;
+import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryColumns;
+
+/**
+ * Class used for reading and writing {@link BatchInformation}.
+ *
+ */
+public class BatchInformationDAO {
+
+ /**
+ * Adds BatchInformation to the {@link FluoQueryColumns#BATCH_COLUMN}.
+ * @param tx - Fluo Transaction
+ * @param nodeId - query node that batch task will be performed on
+ * @param batch - BatchInformation to be processed
+ */
+ public static void addBatch(TransactionBase tx, String nodeId, BatchInformation batch) {
+ Bytes row = BatchRowKeyUtil.getRow(nodeId);
+ tx.set(row, FluoQueryColumns.BATCH_COLUMN, Bytes.of(BatchInformationSerializer.toBytes(batch)));
+ }
+
+ /**
+ * Retrieve BatchInformation
+ * @param tx - Fluo transaction
+ * @param row - row that contains batch information - this is the query id that batch task will be performed on
+ * @return Optional contained the BatchInformation if it is there
+ */
+ public static Optional<BatchInformation> getBatchInformation(TransactionBase tx, Bytes row) {
+ Bytes val = tx.get(row, FluoQueryColumns.BATCH_COLUMN);
+ if(val != null) {
+ return BatchInformationSerializer.fromBytes(val.toArray());
+ } else {
+ return Optional.empty();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/2ca85427/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BatchObserver.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BatchObserver.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BatchObserver.java
new file mode 100644
index 0000000..6194236
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/batch/BatchObserver.java
@@ -0,0 +1,63 @@
+package org.apache.rya.indexing.pcj.fluo.app.batch;
+/*
+ * 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.
+ */
+import java.util.Optional;
+
+import org.apache.fluo.api.client.TransactionBase;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.observer.AbstractObserver;
+import org.apache.rya.indexing.pcj.fluo.app.NodeType;
+import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryColumns;
+
+/**
+ * BatchObserver processes tasks that need to be broken into batches. Entries
+ * stored stored in this {@link FluoQueryColumns#BATCH_COLUMN} are of the form
+ * Row: nodeId, Value: BatchInformation. The nodeId indicates the node that the
+ * batch operation will be performed on. All batch operations are performed on
+ * the bindingSet column for the {@link NodeType} corresponding to the given
+ * nodeId. For example, if the nodeId indicated that the NodeType was
+ * StatementPattern, then the batch operation would be performed on
+ * {@link FluoQueryColumns#STATEMENT_PATTERN_BINDING_SET}. This Observer applies
+ * batch updates to overcome the restriction that all Fluo transactions are processed
+ * in memory. This allows the Rya Fluo application to process large tasks that cannot
+ * fit into memory in a piece-wise, batch fashion.
+ */
+public class BatchObserver extends AbstractObserver {
+
+ /**
+ * Processes the BatchInformation objects when they're written to the Batch column
+ * @param tx - Fluo transaction
+ * @param row - row that contains {@link BatchInformation}
+ * @param col - column that contains BatchInformation
+ */
+ @Override
+ public void process(TransactionBase tx, Bytes row, Column col) throws Exception {
+ Optional<BatchInformation> batchInfo = BatchInformationDAO.getBatchInformation(tx, row);
+ if(batchInfo.isPresent()) {
+ batchInfo.get().getBatchUpdater().processBatch(tx, row, batchInfo.get());
+ }
+ }
+
+ @Override
+ public ObservedColumn getObservedColumn() {
+ return new ObservedColumn(FluoQueryColumns.BATCH_COLUMN, NotificationType.STRONG);
+ }
+
+}