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/04/24 15:06:26 UTC
[8/9] incubator-rya git commit: RYA-260 Fluo PCJ application has had
Aggregation support added to it. Also fixed a bunch of resource leaks that
were causing integration tests to fail. Closes #156.
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c941aea8/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/AccumuloPcjSerializerTest.java
----------------------------------------------------------------------
diff --git a/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/AccumuloPcjSerializerTest.java b/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/AccumuloPcjSerializerTest.java
new file mode 100644
index 0000000..d904d83
--- /dev/null
+++ b/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/AccumuloPcjSerializerTest.java
@@ -0,0 +1,185 @@
+/*
+ * 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;
+
+import static org.junit.Assert.assertEquals;
+
+import org.apache.rya.indexing.pcj.storage.accumulo.AccumuloPcjSerializer;
+import org.apache.rya.indexing.pcj.storage.accumulo.BindingSetConverter;
+import org.apache.rya.indexing.pcj.storage.accumulo.BindingSetConverter.BindingSetConversionException;
+import org.apache.rya.indexing.pcj.storage.accumulo.VariableOrder;
+import org.junit.Test;
+import org.openrdf.model.impl.LiteralImpl;
+import org.openrdf.model.impl.URIImpl;
+import org.openrdf.query.BindingSet;
+import org.openrdf.query.algebra.evaluation.QueryBindingSet;
+import org.openrdf.query.impl.MapBindingSet;
+
+import org.apache.rya.api.resolver.RyaTypeResolverException;
+
+/**
+ * Tests the methods of {@link AccumuloPcjSerialzer}.
+ */
+public class AccumuloPcjSerializerTest {
+
+ /**
+ * The BindingSet has fewer Bindings than there are variables in the variable
+ * order, but they are all in the variable order. This is the case where
+ * the missing bindings were optional.
+ */
+ @Test
+ public void serialize_bindingsSubsetOfVarOrder() throws BindingSetConversionException {
+ // Setup the Binding Set.
+ final MapBindingSet originalBindingSet = new MapBindingSet();
+ originalBindingSet.addBinding("x", new URIImpl("http://a"));
+ originalBindingSet.addBinding("y", new URIImpl("http://b"));
+
+ // Setup the variable order.
+ final VariableOrder varOrder = new VariableOrder("x", "a", "y", "b");
+
+ // Create the byte[] representation of the BindingSet.
+ BindingSetConverter<byte[]> converter = new AccumuloPcjSerializer();
+ byte[] serialized = converter.convert(originalBindingSet, varOrder);
+
+ // Deserialize the byte[] back into the binding set.
+ BindingSet deserialized = converter.convert(serialized, varOrder);
+
+ // Ensure the deserialized value matches the original.
+ assertEquals(originalBindingSet, deserialized);
+ }
+
+ /**
+ * The BindingSet has more Bindings than there are variables in the variable order.
+ * This is the case where a Group By clause does not include all of the Bindings that
+ * are in the Binding Set.
+ */
+ @Test
+ public void serialize_bindingNotInVariableOrder() throws RyaTypeResolverException, BindingSetConversionException {
+ // Setup the Binding Set.
+ final MapBindingSet originalBindingSet = new MapBindingSet();
+ originalBindingSet.addBinding("x", new URIImpl("http://a"));
+ originalBindingSet.addBinding("y", new URIImpl("http://b"));
+ originalBindingSet.addBinding("z", new URIImpl("http://d"));
+
+ // Setup the variable order.
+ final VariableOrder varOrder = new VariableOrder("x", "y");
+
+ // Serialize the Binding Set.
+ BindingSetConverter<byte[]> converter = new AccumuloPcjSerializer();
+ byte[] serialized = converter.convert(originalBindingSet, varOrder);
+
+ // Deserialize it again.
+ BindingSet deserialized = converter.convert(serialized, varOrder);
+
+ // Show that it only contains the bindings that were part of the Variable Order.
+ MapBindingSet expected = new MapBindingSet();
+ expected.addBinding("x", new URIImpl("http://a"));
+ expected.addBinding("y", new URIImpl("http://b"));
+
+ assertEquals(expected, deserialized);
+ }
+
+ @Test
+ public void basicShortUriBsTest() throws BindingSetConversionException {
+ final QueryBindingSet bs = new QueryBindingSet();
+ bs.addBinding("X",new URIImpl("http://uri1"));
+ bs.addBinding("Y",new URIImpl("http://uri2"));
+ final VariableOrder varOrder = new VariableOrder("X","Y");
+
+ BindingSetConverter<byte[]> converter = new AccumuloPcjSerializer();
+ final byte[] byteVal = converter.convert(bs, varOrder);
+ final BindingSet newBs = converter.convert(byteVal, varOrder);
+ assertEquals(bs, newBs);
+ }
+
+ @Test
+ public void basicLongUriBsTest() throws BindingSetConversionException {
+ final QueryBindingSet bs = new QueryBindingSet();
+ bs.addBinding("X",new URIImpl("http://uri1"));
+ bs.addBinding("Y",new URIImpl("http://uri2"));
+ bs.addBinding("Z",new URIImpl("http://uri3"));
+ bs.addBinding("A",new URIImpl("http://uri4"));
+ bs.addBinding("B",new URIImpl("http://uri5"));
+ final VariableOrder varOrder = new VariableOrder("X","Y","Z","A","B");
+
+ BindingSetConverter<byte[]> converter = new AccumuloPcjSerializer();
+ final byte[] byteVal = converter.convert(bs, varOrder);
+ final BindingSet newBs = converter.convert(byteVal, varOrder);
+ assertEquals(bs, newBs);
+ }
+
+ @Test
+ public void basicShortStringLiteralBsTest() throws BindingSetConversionException {
+ final QueryBindingSet bs = new QueryBindingSet();
+ bs.addBinding("X",new LiteralImpl("literal1"));
+ bs.addBinding("Y",new LiteralImpl("literal2"));
+ final VariableOrder varOrder = new VariableOrder("X","Y");
+
+ BindingSetConverter<byte[]> converter = new AccumuloPcjSerializer();
+ final byte[] byteVal = converter.convert(bs, varOrder);
+ final BindingSet newBs = converter.convert(byteVal, varOrder);
+ assertEquals(bs, newBs);
+ }
+
+ @Test
+ public void basicShortMixLiteralBsTest() throws BindingSetConversionException {
+ final QueryBindingSet bs = new QueryBindingSet();
+ bs.addBinding("X",new LiteralImpl("literal1"));
+ bs.addBinding("Y",new LiteralImpl("5", new URIImpl("http://www.w3.org/2001/XMLSchema#integer")));
+ final VariableOrder varOrder = new VariableOrder("X","Y");
+
+ BindingSetConverter<byte[]> converter = new AccumuloPcjSerializer();
+ final byte[] byteVal = converter.convert(bs, varOrder);
+ final BindingSet newBs = converter.convert(byteVal, varOrder);
+ assertEquals(bs, newBs);
+ }
+
+ @Test
+ public void basicLongMixLiteralBsTest() throws BindingSetConversionException {
+ final QueryBindingSet bs = new QueryBindingSet();
+ bs.addBinding("X",new LiteralImpl("literal1"));
+ bs.addBinding("Y",new LiteralImpl("5", new URIImpl("http://www.w3.org/2001/XMLSchema#integer")));
+ bs.addBinding("Z",new LiteralImpl("5.0", new URIImpl("http://www.w3.org/2001/XMLSchema#double")));
+ bs.addBinding("W",new LiteralImpl("1000", new URIImpl("http://www.w3.org/2001/XMLSchema#long")));
+ final VariableOrder varOrder = new VariableOrder("W","X","Y","Z");
+
+ BindingSetConverter<byte[]> converter = new AccumuloPcjSerializer();
+ final byte[] byteVal = converter.convert(bs, varOrder);
+ final BindingSet newBs = converter.convert(byteVal, varOrder);
+ assertEquals(bs, newBs);
+ }
+
+ @Test
+ public void basicMixUriLiteralBsTest() throws BindingSetConversionException {
+ final QueryBindingSet bs = new QueryBindingSet();
+ bs.addBinding("X",new LiteralImpl("literal1"));
+ bs.addBinding("Y",new LiteralImpl("5", new URIImpl("http://www.w3.org/2001/XMLSchema#integer")));
+ bs.addBinding("Z",new LiteralImpl("5.0", new URIImpl("http://www.w3.org/2001/XMLSchema#double")));
+ bs.addBinding("W",new LiteralImpl("1000", new URIImpl("http://www.w3.org/2001/XMLSchema#long")));
+ bs.addBinding("A",new URIImpl("http://uri1"));
+ bs.addBinding("B",new URIImpl("http://uri2"));
+ bs.addBinding("C",new URIImpl("http://uri3"));
+ final VariableOrder varOrder = new VariableOrder("A","W","X","Y","Z","B","C");
+
+ BindingSetConverter<byte[]> converter = new AccumuloPcjSerializer();
+ final byte[] byteVal = converter.convert(bs, varOrder);
+ final BindingSet newBs = converter.convert(byteVal, varOrder);
+ assertEquals(bs, newBs);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c941aea8/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/AccumuloPcjSerialzerTest.java
----------------------------------------------------------------------
diff --git a/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/AccumuloPcjSerialzerTest.java b/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/AccumuloPcjSerialzerTest.java
deleted file mode 100644
index d69205e..0000000
--- a/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/AccumuloPcjSerialzerTest.java
+++ /dev/null
@@ -1,175 +0,0 @@
-package org.apache.rya.indexing.pcj.storage.accumulo;
-
-import static org.junit.Assert.assertEquals;
-
-import org.apache.rya.indexing.pcj.storage.accumulo.AccumuloPcjSerializer;
-import org.apache.rya.indexing.pcj.storage.accumulo.BindingSetConverter;
-import org.apache.rya.indexing.pcj.storage.accumulo.BindingSetConverter.BindingSetConversionException;
-import org.apache.rya.indexing.pcj.storage.accumulo.VariableOrder;
-import org.junit.Test;
-import org.openrdf.model.impl.LiteralImpl;
-import org.openrdf.model.impl.URIImpl;
-import org.openrdf.query.BindingSet;
-import org.openrdf.query.algebra.evaluation.QueryBindingSet;
-import org.openrdf.query.impl.MapBindingSet;
-
-/*
- * 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.rya.api.resolver.RyaTypeResolverException;
-
-/**
- * Tests the methods of {@link AccumuloPcjSerialzer}.
- */
-public class AccumuloPcjSerialzerTest {
-
- /**
- * The BindingSet has fewer Bindings than there are variables in the variable
- * order, but they are all in the variable order. This is the case where
- * the missing bindings were optional.
- */
- @Test
- public void serialize_bindingsSubsetOfVarOrder() throws BindingSetConversionException {
- // Setup the Binding Set.
- final MapBindingSet originalBindingSet = new MapBindingSet();
- originalBindingSet.addBinding("x", new URIImpl("http://a"));
- originalBindingSet.addBinding("y", new URIImpl("http://b"));
-
- // Setup the variable order.
- final VariableOrder varOrder = new VariableOrder("x", "a", "y", "b");
-
- // Create the byte[] representation of the BindingSet.
- BindingSetConverter<byte[]> converter = new AccumuloPcjSerializer();
- byte[] serialized = converter.convert(originalBindingSet, varOrder);
-
- // Deserialize the byte[] back into the binding set.
- BindingSet deserialized = converter.convert(serialized, varOrder);
-
- // Ensure the deserialized value matches the original.
- assertEquals(originalBindingSet, deserialized);
- }
-
- /**
- * The BindingSet has a Binding whose name is not in the variable order.
- * This is illegal.
- */
- @Test(expected = IllegalArgumentException.class)
- public void serialize_bindingNotInVariableOrder() throws RyaTypeResolverException, BindingSetConversionException {
- // Setup the Binding Set.
- final MapBindingSet originalBindingSet = new MapBindingSet();
- originalBindingSet.addBinding("x", new URIImpl("http://a"));
- originalBindingSet.addBinding("y", new URIImpl("http://b"));
- originalBindingSet.addBinding("z", new URIImpl("http://d"));
-
- // Setup the variable order.
- final VariableOrder varOrder = new VariableOrder("x", "y");
-
- // Create the byte[] representation of the BindingSet. This will throw an exception.
- BindingSetConverter<byte[]> converter = new AccumuloPcjSerializer();
- converter.convert(originalBindingSet, varOrder);
- }
-
- @Test
- public void basicShortUriBsTest() throws BindingSetConversionException {
- final QueryBindingSet bs = new QueryBindingSet();
- bs.addBinding("X",new URIImpl("http://uri1"));
- bs.addBinding("Y",new URIImpl("http://uri2"));
- final VariableOrder varOrder = new VariableOrder("X","Y");
-
- BindingSetConverter<byte[]> converter = new AccumuloPcjSerializer();
- final byte[] byteVal = converter.convert(bs, varOrder);
- final BindingSet newBs = converter.convert(byteVal, varOrder);
- assertEquals(bs, newBs);
- }
-
- @Test
- public void basicLongUriBsTest() throws BindingSetConversionException {
- final QueryBindingSet bs = new QueryBindingSet();
- bs.addBinding("X",new URIImpl("http://uri1"));
- bs.addBinding("Y",new URIImpl("http://uri2"));
- bs.addBinding("Z",new URIImpl("http://uri3"));
- bs.addBinding("A",new URIImpl("http://uri4"));
- bs.addBinding("B",new URIImpl("http://uri5"));
- final VariableOrder varOrder = new VariableOrder("X","Y","Z","A","B");
-
- BindingSetConverter<byte[]> converter = new AccumuloPcjSerializer();
- final byte[] byteVal = converter.convert(bs, varOrder);
- final BindingSet newBs = converter.convert(byteVal, varOrder);
- assertEquals(bs, newBs);
- }
-
- @Test
- public void basicShortStringLiteralBsTest() throws BindingSetConversionException {
- final QueryBindingSet bs = new QueryBindingSet();
- bs.addBinding("X",new LiteralImpl("literal1"));
- bs.addBinding("Y",new LiteralImpl("literal2"));
- final VariableOrder varOrder = new VariableOrder("X","Y");
-
- BindingSetConverter<byte[]> converter = new AccumuloPcjSerializer();
- final byte[] byteVal = converter.convert(bs, varOrder);
- final BindingSet newBs = converter.convert(byteVal, varOrder);
- assertEquals(bs, newBs);
- }
-
- @Test
- public void basicShortMixLiteralBsTest() throws BindingSetConversionException {
- final QueryBindingSet bs = new QueryBindingSet();
- bs.addBinding("X",new LiteralImpl("literal1"));
- bs.addBinding("Y",new LiteralImpl("5", new URIImpl("http://www.w3.org/2001/XMLSchema#integer")));
- final VariableOrder varOrder = new VariableOrder("X","Y");
-
- BindingSetConverter<byte[]> converter = new AccumuloPcjSerializer();
- final byte[] byteVal = converter.convert(bs, varOrder);
- final BindingSet newBs = converter.convert(byteVal, varOrder);
- assertEquals(bs, newBs);
- }
-
- @Test
- public void basicLongMixLiteralBsTest() throws BindingSetConversionException {
- final QueryBindingSet bs = new QueryBindingSet();
- bs.addBinding("X",new LiteralImpl("literal1"));
- bs.addBinding("Y",new LiteralImpl("5", new URIImpl("http://www.w3.org/2001/XMLSchema#integer")));
- bs.addBinding("Z",new LiteralImpl("5.0", new URIImpl("http://www.w3.org/2001/XMLSchema#double")));
- bs.addBinding("W",new LiteralImpl("1000", new URIImpl("http://www.w3.org/2001/XMLSchema#long")));
- final VariableOrder varOrder = new VariableOrder("W","X","Y","Z");
-
- BindingSetConverter<byte[]> converter = new AccumuloPcjSerializer();
- final byte[] byteVal = converter.convert(bs, varOrder);
- final BindingSet newBs = converter.convert(byteVal, varOrder);
- assertEquals(bs, newBs);
- }
-
- @Test
- public void basicMixUriLiteralBsTest() throws BindingSetConversionException {
- final QueryBindingSet bs = new QueryBindingSet();
- bs.addBinding("X",new LiteralImpl("literal1"));
- bs.addBinding("Y",new LiteralImpl("5", new URIImpl("http://www.w3.org/2001/XMLSchema#integer")));
- bs.addBinding("Z",new LiteralImpl("5.0", new URIImpl("http://www.w3.org/2001/XMLSchema#double")));
- bs.addBinding("W",new LiteralImpl("1000", new URIImpl("http://www.w3.org/2001/XMLSchema#long")));
- bs.addBinding("A",new URIImpl("http://uri1"));
- bs.addBinding("B",new URIImpl("http://uri2"));
- bs.addBinding("C",new URIImpl("http://uri3"));
- final VariableOrder varOrder = new VariableOrder("A","W","X","Y","Z","B","C");
-
- BindingSetConverter<byte[]> converter = new AccumuloPcjSerializer();
- final byte[] byteVal = converter.convert(bs, varOrder);
- final BindingSet newBs = converter.convert(byteVal, varOrder);
- assertEquals(bs, newBs);
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c941aea8/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/BindingSetStringConverterTest.java
----------------------------------------------------------------------
diff --git a/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/BindingSetStringConverterTest.java b/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/BindingSetStringConverterTest.java
index e01e7de..b263038 100644
--- a/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/BindingSetStringConverterTest.java
+++ b/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/BindingSetStringConverterTest.java
@@ -23,10 +23,7 @@ import static org.junit.Assert.assertEquals;
import java.math.BigDecimal;
import java.math.BigInteger;
-import org.apache.rya.indexing.pcj.storage.accumulo.BindingSetConverter;
import org.apache.rya.indexing.pcj.storage.accumulo.BindingSetConverter.BindingSetConversionException;
-import org.apache.rya.indexing.pcj.storage.accumulo.BindingSetStringConverter;
-import org.apache.rya.indexing.pcj.storage.accumulo.VariableOrder;
import org.junit.Test;
import org.openrdf.model.impl.BooleanLiteralImpl;
import org.openrdf.model.impl.DecimalLiteralImpl;
@@ -41,6 +38,23 @@ import org.openrdf.query.impl.MapBindingSet;
public class BindingSetStringConverterTest {
@Test
+ public void noBindings() throws BindingSetConversionException {
+ // Create a BindingSet that doesn't have any bindings.
+ final MapBindingSet original = new MapBindingSet();
+
+ // Convert it to a String.
+ final VariableOrder varOrder = new VariableOrder();
+ final BindingSetConverter<String> converter = new BindingSetStringConverter();
+ final String bindingSetString = converter.convert(original, varOrder);
+
+ // Convert it back to a binding set.
+ final BindingSet converted = converter.convert(bindingSetString, varOrder);
+
+ // Ensure it is still an empty BindingSet.
+ assertEquals(original, converted);
+ }
+
+ @Test
public void toString_URIs() throws BindingSetConversionException {
// Setup the binding set that will be converted.
final MapBindingSet originalBindingSet = new MapBindingSet();
@@ -53,7 +67,7 @@ public class BindingSetStringConverterTest {
final BindingSetConverter<String> converter = new BindingSetStringConverter();
final String bindingSetString = converter.convert(originalBindingSet, varOrder);
- // Ensure it converted to the expected result.l
+ // Ensure it converted to the expected result.
final String expected =
"http://b<<~>>http://www.w3.org/2001/XMLSchema#anyURI:::" +
"http://c<<~>>http://www.w3.org/2001/XMLSchema#anyURI:::" +
@@ -163,26 +177,6 @@ public class BindingSetStringConverterTest {
assertEquals(expected, bindingSetString);
}
- /**
- * The BindingSet has a Binding whose name is not in the variable order.
- * This is illegal.
- */
- @Test(expected = IllegalArgumentException.class)
- public void toString_bindingNotInVariableOrder() throws BindingSetConversionException {
- // Setup the Binding Set.
- final MapBindingSet originalBindingSet = new MapBindingSet();
- originalBindingSet.addBinding("x", new URIImpl("http://a"));
- originalBindingSet.addBinding("y", new URIImpl("http://b"));
- originalBindingSet.addBinding("z", new URIImpl("http://d"));
-
- // Setup the variable order.
- final VariableOrder varOrder = new VariableOrder("x", "y");
-
- // Create the String representation of the BindingSet. This will throw an exception.
- final BindingSetConverter<String> converter = new BindingSetStringConverter();
- converter.convert(originalBindingSet, varOrder);
- }
-
@Test
public void fromString() throws BindingSetConversionException {
// Setup the String that will be converted.
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c941aea8/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/PcjTablesIntegrationTest.java
----------------------------------------------------------------------
diff --git a/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/PcjTablesIntegrationTest.java b/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/PcjTablesIntegrationTest.java
index 623526e..2bcce65 100644
--- a/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/PcjTablesIntegrationTest.java
+++ b/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/PcjTablesIntegrationTest.java
@@ -38,10 +38,17 @@ import org.apache.accumulo.minicluster.MiniAccumuloCluster;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
+import org.apache.rya.accumulo.AccumuloRdfConfiguration;
+import org.apache.rya.accumulo.AccumuloRyaDAO;
+import org.apache.rya.accumulo.MiniAccumuloClusterInstance;
+import org.apache.rya.api.RdfCloudTripleStoreConfiguration;
import org.apache.rya.indexing.pcj.storage.PcjException;
import org.apache.rya.indexing.pcj.storage.PcjMetadata;
+import org.apache.rya.indexing.pcj.storage.PrecomputedJoinStorage.CloseableIterator;
import org.apache.rya.indexing.pcj.storage.PrecomputedJoinStorage.PCJStorageException;
import org.apache.rya.indexing.pcj.storage.accumulo.BindingSetConverter.BindingSetConversionException;
+import org.apache.rya.rdftriplestore.RdfCloudTripleStore;
+import org.apache.rya.rdftriplestore.RyaSailRepository;
import org.apache.zookeeper.ClientCnxn;
import org.junit.After;
import org.junit.Before;
@@ -63,13 +70,6 @@ import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
-import org.apache.rya.accumulo.AccumuloRdfConfiguration;
-import org.apache.rya.accumulo.AccumuloRyaDAO;
-import org.apache.rya.accumulo.MiniAccumuloClusterInstance;
-import org.apache.rya.api.RdfCloudTripleStoreConfiguration;
-import org.apache.rya.rdftriplestore.RdfCloudTripleStore;
-import org.apache.rya.rdftriplestore.RyaSailRepository;
-
/**
* Performs integration test using {@link MiniAccumuloCluster} to ensure the
* functions of {@link PcjTables} work within a cluster setting.
@@ -237,7 +237,7 @@ public class PcjTablesIntegrationTest {
}
@Test
- public void listResults() throws PCJStorageException, AccumuloException, AccumuloSecurityException {
+ public void listResults() throws Exception {
final String sparql =
"SELECT ?name ?age " +
"{" +
@@ -274,8 +274,14 @@ public class PcjTablesIntegrationTest {
// Fetch the Binding Sets that have been stored in the PCJ table.
final Set<BindingSet> results = new HashSet<>();
- for(final BindingSet result : pcjs.listResults(accumuloConn, pcjTableName, new Authorizations())) {
- results.add( result );
+
+ final CloseableIterator<BindingSet> resultsIt = pcjs.listResults(accumuloConn, pcjTableName, new Authorizations());
+ try {
+ while(resultsIt.hasNext()) {
+ results.add( resultsIt.next() );
+ }
+ } finally {
+ resultsIt.close();
}
// Verify the fetched results match the expected ones.
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c941aea8/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/accumulo/AccumuloPcjStorageIT.java
----------------------------------------------------------------------
diff --git a/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/accumulo/AccumuloPcjStorageIT.java b/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/accumulo/AccumuloPcjStorageIT.java
index 7eeff1d..98ed4c7 100644
--- a/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/accumulo/AccumuloPcjStorageIT.java
+++ b/extras/rya.indexing.pcj/src/test/java/org/apache/rya/indexing/pcj/storage/accumulo/accumulo/AccumuloPcjStorageIT.java
@@ -31,8 +31,15 @@ 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.rya.accumulo.AccumuloRyaITBase;
+import org.apache.rya.accumulo.instance.AccumuloRyaInstanceDetailsRepository;
+import org.apache.rya.api.instance.RyaDetails.PCJIndexDetails.PCJDetails;
+import org.apache.rya.api.instance.RyaDetailsRepository;
+import org.apache.rya.api.instance.RyaDetailsRepository.NotInitializedException;
+import org.apache.rya.api.instance.RyaDetailsRepository.RyaDetailsRepositoryException;
import org.apache.rya.indexing.pcj.storage.PcjMetadata;
import org.apache.rya.indexing.pcj.storage.PrecomputedJoinStorage;
+import org.apache.rya.indexing.pcj.storage.PrecomputedJoinStorage.CloseableIterator;
import org.apache.rya.indexing.pcj.storage.PrecomputedJoinStorage.PCJStorageException;
import org.apache.rya.indexing.pcj.storage.accumulo.AccumuloPcjStorage;
import org.apache.rya.indexing.pcj.storage.accumulo.ShiftVarOrderFactory;
@@ -46,13 +53,6 @@ import org.openrdf.query.impl.MapBindingSet;
import com.google.common.collect.ImmutableMap;
-import org.apache.rya.accumulo.AccumuloRyaITBase;
-import org.apache.rya.accumulo.instance.AccumuloRyaInstanceDetailsRepository;
-import org.apache.rya.api.instance.RyaDetails.PCJIndexDetails.PCJDetails;
-import org.apache.rya.api.instance.RyaDetailsRepository;
-import org.apache.rya.api.instance.RyaDetailsRepository.NotInitializedException;
-import org.apache.rya.api.instance.RyaDetailsRepository.RyaDetailsRepositoryException;
-
/**
* Integration tests the methods of {@link AccumuloPcjStorage}.
* </p>
@@ -66,23 +66,23 @@ public class AccumuloPcjStorageIT extends AccumuloRyaITBase {
// Setup the PCJ storage that will be tested against.
final Connector connector = super.getClusterInstance().getConnector();
final String ryaInstanceName = super.getRyaInstanceName();
- final PrecomputedJoinStorage pcjStorage = new AccumuloPcjStorage(connector, ryaInstanceName);
-
- // Create a PCJ.
- final String pcjId = pcjStorage.createPcj("SELECT * WHERE { ?a <http://isA> ?b } ");
+ try(final PrecomputedJoinStorage pcjStorage = new AccumuloPcjStorage(connector, ryaInstanceName)) {
+ // Create a PCJ.
+ final String pcjId = pcjStorage.createPcj("SELECT * WHERE { ?a <http://isA> ?b } ");
- // Ensure the Rya details have been updated to include the PCJ's ID.
- final RyaDetailsRepository detailsRepo = new AccumuloRyaInstanceDetailsRepository(connector, ryaInstanceName);
+ // Ensure the Rya details have been updated to include the PCJ's ID.
+ final RyaDetailsRepository detailsRepo = new AccumuloRyaInstanceDetailsRepository(connector, ryaInstanceName);
- final ImmutableMap<String, PCJDetails> detailsMap = detailsRepo.getRyaInstanceDetails()
- .getPCJIndexDetails()
- .getPCJDetails();
+ final ImmutableMap<String, PCJDetails> detailsMap = detailsRepo.getRyaInstanceDetails()
+ .getPCJIndexDetails()
+ .getPCJDetails();
- final PCJDetails expectedDetails = PCJDetails.builder()
- .setId( pcjId )
- .build();
+ final PCJDetails expectedDetails = PCJDetails.builder()
+ .setId( pcjId )
+ .build();
- assertEquals(expectedDetails, detailsMap.get(pcjId));
+ assertEquals(expectedDetails, detailsMap.get(pcjId));
+ }
}
@Test
@@ -90,22 +90,22 @@ public class AccumuloPcjStorageIT extends AccumuloRyaITBase {
// Setup the PCJ storage that will be tested against.
final Connector connector = super.getClusterInstance().getConnector();
final String ryaInstanceName = super.getRyaInstanceName();
- final PrecomputedJoinStorage pcjStorage = new AccumuloPcjStorage(connector, ryaInstanceName);
-
- // Create a PCJ.
- final String pcjId = pcjStorage.createPcj("SELECT * WHERE { ?a <http://isA> ?b } ");
+ try(final PrecomputedJoinStorage pcjStorage = new AccumuloPcjStorage(connector, ryaInstanceName)) {
+ // Create a PCJ.
+ final String pcjId = pcjStorage.createPcj("SELECT * WHERE { ?a <http://isA> ?b } ");
- // Delete the PCJ that was just created.
- pcjStorage.dropPcj(pcjId);
+ // Delete the PCJ that was just created.
+ pcjStorage.dropPcj(pcjId);
- // Ensure the Rya details have been updated to no longer include the PCJ's ID.
- final RyaDetailsRepository detailsRepo = new AccumuloRyaInstanceDetailsRepository(connector, ryaInstanceName);
+ // Ensure the Rya details have been updated to no longer include the PCJ's ID.
+ final RyaDetailsRepository detailsRepo = new AccumuloRyaInstanceDetailsRepository(connector, ryaInstanceName);
- final ImmutableMap<String, PCJDetails> detailsMap = detailsRepo.getRyaInstanceDetails()
- .getPCJIndexDetails()
- .getPCJDetails();
+ final ImmutableMap<String, PCJDetails> detailsMap = detailsRepo.getRyaInstanceDetails()
+ .getPCJIndexDetails()
+ .getPCJDetails();
- assertFalse( detailsMap.containsKey(pcjId) );
+ assertFalse( detailsMap.containsKey(pcjId) );
+ }
}
@Test
@@ -113,27 +113,27 @@ public class AccumuloPcjStorageIT extends AccumuloRyaITBase {
// Setup the PCJ storage that will be tested against.
final Connector connector = super.getClusterInstance().getConnector();
final String ryaInstanceName = super.getRyaInstanceName();
- final PrecomputedJoinStorage pcjStorage = new AccumuloPcjStorage(connector, ryaInstanceName);
-
- // Create a few PCJs and hold onto their IDs.
- final List<String> expectedIds = new ArrayList<>();
+ try(final PrecomputedJoinStorage pcjStorage = new AccumuloPcjStorage(connector, ryaInstanceName)) {
+ // Create a few PCJs and hold onto their IDs.
+ final List<String> expectedIds = new ArrayList<>();
- String pcjId = pcjStorage.createPcj("SELECT * WHERE { ?a <http://isA> ?b } ");
- expectedIds.add( pcjId );
+ String pcjId = pcjStorage.createPcj("SELECT * WHERE { ?a <http://isA> ?b } ");
+ expectedIds.add( pcjId );
- pcjId = pcjStorage.createPcj("SELECT * WHERE { ?a <http://isA> ?b } ");
- expectedIds.add( pcjId );
+ pcjId = pcjStorage.createPcj("SELECT * WHERE { ?a <http://isA> ?b } ");
+ expectedIds.add( pcjId );
- pcjId = pcjStorage.createPcj("SELECT * WHERE { ?a <http://isA> ?b } ");
- expectedIds.add( pcjId );
+ pcjId = pcjStorage.createPcj("SELECT * WHERE { ?a <http://isA> ?b } ");
+ expectedIds.add( pcjId );
- // Fetch the PCJ names
- final List<String> pcjIds = pcjStorage.listPcjs();
+ // Fetch the PCJ names
+ final List<String> pcjIds = pcjStorage.listPcjs();
- // Ensure the expected IDs match the fetched IDs.
- Collections.sort(expectedIds);
- Collections.sort(pcjIds);
- assertEquals(expectedIds, pcjIds);
+ // Ensure the expected IDs match the fetched IDs.
+ Collections.sort(expectedIds);
+ Collections.sort(pcjIds);
+ assertEquals(expectedIds, pcjIds);
+ }
}
@Test
@@ -141,19 +141,19 @@ public class AccumuloPcjStorageIT extends AccumuloRyaITBase {
// Setup the PCJ storage that will be tested against.
final Connector connector = super.getClusterInstance().getConnector();
final String ryaInstanceName = super.getRyaInstanceName();
- final PrecomputedJoinStorage pcjStorage = new AccumuloPcjStorage(connector, ryaInstanceName);
-
- // Create a PCJ.
- final String sparql = "SELECT * WHERE { ?a <http://isA> ?b }";
- final String pcjId = pcjStorage.createPcj(sparql);
-
- // Fetch the PCJ's metadata.
- final PcjMetadata metadata = pcjStorage.getPcjMetadata(pcjId);
-
- // Ensure it has the expected values.
- final Set<VariableOrder> varOrders = new ShiftVarOrderFactory().makeVarOrders(sparql);
- final PcjMetadata expectedMetadata = new PcjMetadata(sparql, 0L, varOrders);
- assertEquals(expectedMetadata, metadata);
+ try(final PrecomputedJoinStorage pcjStorage = new AccumuloPcjStorage(connector, ryaInstanceName)) {
+ // Create a PCJ.
+ final String sparql = "SELECT * WHERE { ?a <http://isA> ?b }";
+ final String pcjId = pcjStorage.createPcj(sparql);
+
+ // Fetch the PCJ's metadata.
+ final PcjMetadata metadata = pcjStorage.getPcjMetadata(pcjId);
+
+ // Ensure it has the expected values.
+ final Set<VariableOrder> varOrders = new ShiftVarOrderFactory().makeVarOrders(sparql);
+ final PcjMetadata expectedMetadata = new PcjMetadata(sparql, 0L, varOrders);
+ assertEquals(expectedMetadata, metadata);
+ }
}
@Test
@@ -161,112 +161,116 @@ public class AccumuloPcjStorageIT extends AccumuloRyaITBase {
// Setup the PCJ storage that will be tested against.
final Connector connector = super.getClusterInstance().getConnector();
final String ryaInstanceName = super.getRyaInstanceName();
- final PrecomputedJoinStorage pcjStorage = new AccumuloPcjStorage(connector, ryaInstanceName);
+ try(final PrecomputedJoinStorage pcjStorage = new AccumuloPcjStorage(connector, ryaInstanceName)) {
+ // Create a PCJ.
+ final String sparql = "SELECT * WHERE { ?a <http://isA> ?b }";
+ final String pcjId = pcjStorage.createPcj(sparql);
- // Create a PCJ.
- final String sparql = "SELECT * WHERE { ?a <http://isA> ?b }";
- final String pcjId = pcjStorage.createPcj(sparql);
+ // Add some binding sets to it.
+ final Set<VisibilityBindingSet> results = new HashSet<>();
- // Add some binding sets to it.
- final Set<VisibilityBindingSet> results = new HashSet<>();
+ final MapBindingSet aliceBS = new MapBindingSet();
+ aliceBS.addBinding("a", new URIImpl("http://Alice"));
+ aliceBS.addBinding("b", new URIImpl("http://Person"));
+ results.add( new VisibilityBindingSet(aliceBS, "") );
- final MapBindingSet aliceBS = new MapBindingSet();
- aliceBS.addBinding("a", new URIImpl("http://Alice"));
- aliceBS.addBinding("b", new URIImpl("http://Person"));
- results.add( new VisibilityBindingSet(aliceBS, "") );
+ final MapBindingSet charlieBS = new MapBindingSet();
+ charlieBS.addBinding("a", new URIImpl("http://Charlie"));
+ charlieBS.addBinding("b", new URIImpl("http://Comedian"));
+ results.add( new VisibilityBindingSet(charlieBS, "") );
- final MapBindingSet charlieBS = new MapBindingSet();
- charlieBS.addBinding("a", new URIImpl("http://Charlie"));
- charlieBS.addBinding("b", new URIImpl("http://Comedian"));
- results.add( new VisibilityBindingSet(charlieBS, "") );
+ pcjStorage.addResults(pcjId, results);
- pcjStorage.addResults(pcjId, results);
+ // Make sure the PCJ metadata was updated.
+ final PcjMetadata metadata = pcjStorage.getPcjMetadata(pcjId);
- // Make sure the PCJ metadata was updated.
- final PcjMetadata metadata = pcjStorage.getPcjMetadata(pcjId);
-
- final Set<VariableOrder> varOrders = new ShiftVarOrderFactory().makeVarOrders(sparql);
- final PcjMetadata expectedMetadata = new PcjMetadata(sparql, 2L, varOrders);
- assertEquals(expectedMetadata, metadata);
+ final Set<VariableOrder> varOrders = new ShiftVarOrderFactory().makeVarOrders(sparql);
+ final PcjMetadata expectedMetadata = new PcjMetadata(sparql, 2L, varOrders);
+ assertEquals(expectedMetadata, metadata);
+ }
}
@Test
- public void listResults() throws AccumuloException, AccumuloSecurityException, PCJStorageException {
+ public void listResults() throws Exception {
// Setup the PCJ storage that will be tested against.
final Connector connector = super.getClusterInstance().getConnector();
final String ryaInstanceName = super.getRyaInstanceName();
- final PrecomputedJoinStorage pcjStorage = new AccumuloPcjStorage(connector, ryaInstanceName);
-
- // Create a PCJ.
- final String sparql = "SELECT * WHERE { ?a <http://isA> ?b }";
- final String pcjId = pcjStorage.createPcj(sparql);
-
- // Add some binding sets to it.
- final Set<VisibilityBindingSet> expectedResults = new HashSet<>();
-
- final MapBindingSet aliceBS = new MapBindingSet();
- aliceBS.addBinding("a", new URIImpl("http://Alice"));
- aliceBS.addBinding("b", new URIImpl("http://Person"));
- expectedResults.add( new VisibilityBindingSet(aliceBS, "") );
-
- final MapBindingSet charlieBS = new MapBindingSet();
- charlieBS.addBinding("a", new URIImpl("http://Charlie"));
- charlieBS.addBinding("b", new URIImpl("http://Comedian"));
- expectedResults.add( new VisibilityBindingSet(charlieBS, "") );
-
- pcjStorage.addResults(pcjId, expectedResults);
-
- // List the results that were stored.
- final Set<BindingSet> results = new HashSet<>();
- for(final BindingSet result : pcjStorage.listResults(pcjId)) {
- results.add( result );
+ try(final PrecomputedJoinStorage pcjStorage = new AccumuloPcjStorage(connector, ryaInstanceName)) {
+ // Create a PCJ.
+ final String sparql = "SELECT * WHERE { ?a <http://isA> ?b }";
+ final String pcjId = pcjStorage.createPcj(sparql);
+
+ // Add some binding sets to it.
+ final Set<VisibilityBindingSet> expectedResults = new HashSet<>();
+
+ final MapBindingSet aliceBS = new MapBindingSet();
+ aliceBS.addBinding("a", new URIImpl("http://Alice"));
+ aliceBS.addBinding("b", new URIImpl("http://Person"));
+ expectedResults.add( new VisibilityBindingSet(aliceBS, "") );
+
+ final MapBindingSet charlieBS = new MapBindingSet();
+ charlieBS.addBinding("a", new URIImpl("http://Charlie"));
+ charlieBS.addBinding("b", new URIImpl("http://Comedian"));
+ expectedResults.add( new VisibilityBindingSet(charlieBS, "") );
+
+ pcjStorage.addResults(pcjId, expectedResults);
+
+ // List the results that were stored.
+ final Set<BindingSet> results = new HashSet<>();
+ try(CloseableIterator<BindingSet> resultsIt = pcjStorage.listResults(pcjId)) {
+ while(resultsIt.hasNext()) {
+ results.add( resultsIt.next() );
+ }
+ }
+
+ assertEquals(expectedResults, results);
}
-
- assertEquals(expectedResults, results);
}
@Test
- public void purge() throws AccumuloException, AccumuloSecurityException, PCJStorageException, MalformedQueryException {
+ public void purge() throws Exception {
// Setup the PCJ storage that will be tested against.
final Connector connector = super.getClusterInstance().getConnector();
final String ryaInstanceName = super.getRyaInstanceName();
- final PrecomputedJoinStorage pcjStorage = new AccumuloPcjStorage(connector, ryaInstanceName);
-
- // Create a PCJ.
- final String sparql = "SELECT * WHERE { ?a <http://isA> ?b }";
- final String pcjId = pcjStorage.createPcj(sparql);
+ try(final PrecomputedJoinStorage pcjStorage = new AccumuloPcjStorage(connector, ryaInstanceName)) {
+ // Create a PCJ.
+ final String sparql = "SELECT * WHERE { ?a <http://isA> ?b }";
+ final String pcjId = pcjStorage.createPcj(sparql);
- // Add some binding sets to it.
- final Set<VisibilityBindingSet> expectedResults = new HashSet<>();
+ // Add some binding sets to it.
+ final Set<VisibilityBindingSet> expectedResults = new HashSet<>();
- final MapBindingSet aliceBS = new MapBindingSet();
- aliceBS.addBinding("a", new URIImpl("http://Alice"));
- aliceBS.addBinding("b", new URIImpl("http://Person"));
- expectedResults.add( new VisibilityBindingSet(aliceBS, "") );
+ final MapBindingSet aliceBS = new MapBindingSet();
+ aliceBS.addBinding("a", new URIImpl("http://Alice"));
+ aliceBS.addBinding("b", new URIImpl("http://Person"));
+ expectedResults.add( new VisibilityBindingSet(aliceBS, "") );
- final MapBindingSet charlieBS = new MapBindingSet();
- charlieBS.addBinding("a", new URIImpl("http://Charlie"));
- charlieBS.addBinding("b", new URIImpl("http://Comedian"));
- expectedResults.add( new VisibilityBindingSet(charlieBS, "") );
+ final MapBindingSet charlieBS = new MapBindingSet();
+ charlieBS.addBinding("a", new URIImpl("http://Charlie"));
+ charlieBS.addBinding("b", new URIImpl("http://Comedian"));
+ expectedResults.add( new VisibilityBindingSet(charlieBS, "") );
- pcjStorage.addResults(pcjId, expectedResults);
+ pcjStorage.addResults(pcjId, expectedResults);
- // Purge the PCJ.
- pcjStorage.purge(pcjId);
+ // Purge the PCJ.
+ pcjStorage.purge(pcjId);
- // List the results that were stored.
- final Set<BindingSet> results = new HashSet<>();
- for(final BindingSet result : pcjStorage.listResults(pcjId)) {
- results.add( result );
- }
+ // List the results that were stored.
+ final Set<BindingSet> results = new HashSet<>();
+ try(CloseableIterator<BindingSet> resultsIt = pcjStorage.listResults(pcjId)) {
+ while(resultsIt.hasNext()) {
+ results.add( resultsIt.next() );
+ }
+ }
- assertTrue( results.isEmpty() );
+ assertTrue( results.isEmpty() );
- // Make sure the PCJ metadata was updated.
- final PcjMetadata metadata = pcjStorage.getPcjMetadata(pcjId);
+ // Make sure the PCJ metadata was updated.
+ final PcjMetadata metadata = pcjStorage.getPcjMetadata(pcjId);
- final Set<VariableOrder> varOrders = new ShiftVarOrderFactory().makeVarOrders(sparql);
- final PcjMetadata expectedMetadata = new PcjMetadata(sparql, 0L, varOrders);
- assertEquals(expectedMetadata, metadata);
+ final Set<VariableOrder> varOrders = new ShiftVarOrderFactory().makeVarOrders(sparql);
+ final PcjMetadata expectedMetadata = new PcjMetadata(sparql, 0L, varOrders);
+ assertEquals(expectedMetadata, metadata);
+ }
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c941aea8/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 86f5b48..1de0813 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
@@ -22,9 +22,9 @@ 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.Iterator;
import java.util.List;
import java.util.Set;
@@ -34,6 +34,7 @@ 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.domain.RyaStatement;
@@ -52,15 +53,14 @@ import org.apache.rya.indexing.pcj.fluo.app.query.StatementPatternMetadata;
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.openrdf.model.Resource;
import org.openrdf.model.URI;
import org.openrdf.model.Value;
import org.openrdf.query.MalformedQueryException;
-import org.openrdf.query.QueryEvaluationException;
import org.openrdf.query.algebra.StatementPattern;
import org.openrdf.query.parser.ParsedQuery;
import org.openrdf.query.parser.sparql.SPARQLParser;
-import org.openrdf.sail.SailException;
import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
import edu.umd.cs.findbugs.annotations.NonNull;
@@ -82,6 +82,7 @@ import edu.umd.cs.findbugs.annotations.NonNull;
*/
@DefaultAnnotation(NonNull.class)
public class CreatePcj {
+ private static final Logger log = Logger.getLogger(CreatePcj.class);
/**
* The default Statement Pattern batch insert size is 1000.
@@ -113,7 +114,51 @@ public class CreatePcj {
this.spInsertBatchSize = spInsertBatchSize;
}
-
+ /**
+ * Tells the Fluo PCJ Updater application to maintain a new PCJ.
+ *
+ * @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);
+
+ // 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);
+
+ try (Transaction tx = fluo.newTransaction()) {
+ // Write the query's structure to Fluo.
+ new FluoQueryMetadataDAO().write(tx, fluoQuery);
+
+ // 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().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.
* <p>
@@ -126,147 +171,115 @@ public class CreatePcj {
* @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 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 SailException Historic PCJ results could not be loaded because of a problem with {@code rya}.
- * @throws QueryEvaluationException Historic PCJ results could not be loaded because of a problem with {@code rya}.
+ * @throws RyaDAOException Historic PCJ results could not be loaded because of a problem with {@code rya}.
*/
- public String withRyaIntegration(final String pcjId, final PrecomputedJoinStorage pcjStorage, final FluoClient fluo,
- final Connector accumulo, String ryaInstance )
- throws MalformedQueryException, PcjException, SailException, QueryEvaluationException, RyaDAOException {
+ public String withRyaIntegration(
+ final String pcjId,
+ final PrecomputedJoinStorage pcjStorage,
+ final FluoClient fluo,
+ final Connector accumulo,
+ final String ryaInstance ) throws MalformedQueryException, PcjException, RyaDAOException {
requireNonNull(pcjId);
requireNonNull(pcjStorage);
requireNonNull(fluo);
- requireNonNull(accumulo);
- requireNonNull(ryaInstance);
-
- //Create AccumuloRyaQueryEngine to query for historic results
- AccumuloRdfConfiguration conf = new AccumuloRdfConfiguration();
- conf.setTablePrefix(ryaInstance);
- conf.setAuths(getAuths(accumulo));
- AccumuloRyaQueryEngine queryEngine = new AccumuloRyaQueryEngine(accumulo, conf);
-
-
- // 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);
+ requireNonNull(accumulo);
+ requireNonNull(ryaInstance);
+
+ // Write the SPARQL query's structure to the Fluo Application.
+ final FluoQuery fluoQuery = createPcj(pcjId, pcjStorage, fluo);
+
+ // 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);
+ }
// return queryId to the caller for later monitoring from the export.
- String queryId = null;
-
- try (Transaction tx = fluo.newTransaction()) {
- // Write the query's structure to Fluo.
- new FluoQueryMetadataDAO().write(tx, fluoQuery);
-
- // The results of the query are eventually exported to an instance
- // of Rya, so store the Rya ID for the PCJ.
- queryId = fluoQuery.getQueryMetadata().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();
- }
-
- // 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));
- }
-
- Iterator<RyaStatement> triples = queryEngine.query(new BatchRyaQuery(queryBatch)).iterator();
- Set<RyaStatement> triplesBatch = new HashSet<>();
-
- // Insert batches of the binding sets into Fluo.
- while (triples.hasNext()) {
- if (triplesBatch.size() == spInsertBatchSize) {
- writeBatch(fluo, triplesBatch);
- triplesBatch.clear();
- }
-
- triplesBatch.add(triples.next());
- }
-
- if (!triplesBatch.isEmpty()) {
- writeBatch(fluo, triplesBatch);
- triplesBatch.clear();
- }
- return queryId;
+ return fluoQuery.getQueryMetadata().getNodeId();
}
-
+
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 URI)) {
+ throw new AssertionError("Predicate must be a URI.");
+ }
+ predURI = RdfToRyaConversions.convertURI((URI) predVal);
+ }
+
+ if (objVal != null ) {
+ objType = RdfToRyaConversions.convertValue(objVal);
+ }
+
+ return new RyaStatement(subjURI, predURI, objType);
}
-
-
- private static RyaStatement spToRyaStatement(StatementPattern sp) {
-
- Value subjVal = sp.getSubjectVar().getValue();
- Value predVal = sp.getPredicateVar().getValue();
- 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 URI)) {
- throw new AssertionError("Predicate must be a URI.");
- }
- predURI = RdfToRyaConversions.convertURI((URI) 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());
+ }
}
-
-
- private String[] getAuths(Connector accumulo) {
- Authorizations auths;
- try {
- auths = accumulo.securityOperations().getUserAuthorizations(accumulo.whoami());
- List<byte[]> authList = auths.getAuthorizations();
- 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());
- }
- }
-
-
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c941aea8/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 1d92262..c11f9fb 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
@@ -25,10 +25,15 @@ import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
-import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
-import edu.umd.cs.findbugs.annotations.NonNull;
-
+import org.apache.fluo.api.client.FluoClient;
+import org.apache.fluo.api.client.Transaction;
+import org.apache.fluo.api.client.scanner.CellScanner;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.RowColumnValue;
+import org.apache.fluo.api.data.Span;
import org.apache.rya.indexing.pcj.fluo.app.NodeType;
+import org.apache.rya.indexing.pcj.fluo.app.query.AggregationMetadata;
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;
@@ -36,13 +41,8 @@ import org.apache.rya.indexing.pcj.fluo.app.query.JoinMetadata;
import org.apache.rya.indexing.pcj.fluo.app.query.QueryMetadata;
import org.openrdf.query.BindingSet;
-import org.apache.fluo.api.client.FluoClient;
-import org.apache.fluo.api.client.Transaction;
-import org.apache.fluo.api.client.scanner.CellScanner;
-import org.apache.fluo.api.data.Bytes;
-import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.data.RowColumnValue;
-import org.apache.fluo.api.data.Span;
+import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
+import edu.umd.cs.findbugs.annotations.NonNull;
/**
* Deletes a Pre-computed Join (PCJ) from Fluo.
@@ -154,6 +154,12 @@ public class DeletePcj {
nodeIds.add(filterChild);
getChildNodeIds(tx, filterChild, nodeIds);
break;
+ case AGGREGATION:
+ final AggregationMetadata aggMeta = dao.readAggregationMetadata(tx, nodeId);
+ final String aggChild = aggMeta.getChildNodeId();
+ nodeIds.add(aggChild);
+ getChildNodeIds(tx, aggChild, nodeIds);
+ break;
case STATEMENT_PATTERN:
break;
}
@@ -254,7 +260,7 @@ public class DeletePcj {
try(Transaction ntx = tx) {
int count = 0;
- Iterator<RowColumnValue> iter = scanner.iterator();
+ final Iterator<RowColumnValue> iter = scanner.iterator();
while (iter.hasNext() && count < batchSize) {
final Bytes row = iter.next().getRow();
count++;
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c941aea8/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 343713c..38fff95 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/pom.xml
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/pom.xml
@@ -64,6 +64,14 @@ under the License.
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.fluo</groupId>
+ <artifactId>fluo-recipes-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.fluo</groupId>
+ <artifactId>fluo-recipes-accumulo</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
@@ -81,7 +89,7 @@ under the License.
</exclusion>
</exclusions>
</dependency>
- <dependency>
+ <dependency>
<groupId>com.esotericsoftware</groupId>
<artifactId>kryo</artifactId>
<version>${kryo.version}</version>