You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rya.apache.org by pu...@apache.org on 2016/05/13 13:13:11 UTC
[1/3] incubator-rya git commit: RYA-53 Added visibility support
Repository: incubator-rya
Updated Branches:
refs/heads/develop 8168b85a7 -> c53b5402b
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/visibility/PcjVisibilityIT.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/visibility/PcjVisibilityIT.java b/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/visibility/PcjVisibilityIT.java
new file mode 100644
index 0000000..07211d5
--- /dev/null
+++ b/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/visibility/PcjVisibilityIT.java
@@ -0,0 +1,362 @@
+/*
+ * 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.visibility;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.admin.SecurityOperations;
+import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.security.TablePermission;
+import org.apache.hadoop.io.Text;
+import org.apache.rya.indexing.pcj.fluo.ITBase;
+import org.apache.rya.indexing.pcj.fluo.api.CreatePcj;
+import org.apache.rya.indexing.pcj.fluo.api.InsertTriples;
+import org.apache.rya.indexing.pcj.fluo.app.export.rya.RyaExportParameters;
+import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryColumns;
+import org.junit.Test;
+import org.openrdf.model.impl.NumericLiteralImpl;
+import org.openrdf.model.impl.URIImpl;
+import org.openrdf.model.vocabulary.XMLSchema;
+import org.openrdf.query.BindingSet;
+import org.openrdf.query.impl.BindingImpl;
+import org.openrdf.query.impl.MapBindingSet;
+import org.openrdf.repository.RepositoryException;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Multimap;
+import com.google.common.collect.Sets;
+
+import io.fluo.api.client.Snapshot;
+import io.fluo.api.data.Bytes;
+import mvm.rya.api.domain.RyaStatement;
+import mvm.rya.api.resolver.RyaTypeResolverException;
+import mvm.rya.indexing.accumulo.VisibilityBindingSet;
+import mvm.rya.indexing.external.tupleSet.AccumuloPcjSerializer;
+import mvm.rya.indexing.external.tupleSet.BindingSetConverter.BindingSetConversionException;
+import mvm.rya.indexing.external.tupleSet.PcjTables;
+import mvm.rya.indexing.external.tupleSet.PcjTables.PcjException;
+import mvm.rya.indexing.external.tupleSet.PcjTables.PcjMetadata;
+import mvm.rya.indexing.external.tupleSet.PcjTables.PcjTableNameFactory;
+import mvm.rya.indexing.external.tupleSet.PcjTables.PcjVarOrderFactory;
+import mvm.rya.indexing.external.tupleSet.PcjTables.ShiftVarOrderFactory;
+import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
+
+public class PcjVisibilityIT extends ITBase {
+
+ private static final AccumuloPcjSerializer converter = new AccumuloPcjSerializer();
+ /**
+ * Configure the export observer to use the Mini Accumulo instance as the
+ * export destination for new PCJ results.
+ */
+ @Override
+ protected Map<String, String> makeExportParams() {
+ final HashMap<String, String> params = new HashMap<>();
+
+ final RyaExportParameters ryaParams = new RyaExportParameters(params);
+ ryaParams.setExportToRya(true);
+ ryaParams.setAccumuloInstanceName(accumulo.getInstanceName());
+ ryaParams.setZookeeperServers(accumulo.getZooKeepers());
+ ryaParams.setExporterUsername("root");
+ ryaParams.setExporterPassword("password");
+
+ return params;
+ }
+
+ @Test
+ public void createWithVisibilityDirect() throws RepositoryException, PcjException, TableNotFoundException, RyaTypeResolverException, AccumuloException, AccumuloSecurityException, BindingSetConversionException {
+ // Create a PCJ table that will include those triples in its results.
+ final String sparql =
+ "SELECT ?name ?age " +
+ "{" +
+ "FILTER(?age < 30) ." +
+ "?name <http://hasAge> ?age." +
+ "?name <http://playsSport> \"Soccer\" " +
+ "}";
+
+ final String pcjTableName = new PcjTableNameFactory().makeTableName(RYA_TABLE_PREFIX, "testPcj");
+
+ // Create and populate the PCJ table.
+ final PcjTables pcjs = new PcjTables();
+ final PcjVarOrderFactory varOrderFactory = Optional.<PcjVarOrderFactory>absent().or(new ShiftVarOrderFactory());
+ final Set<VariableOrder> varOrders = varOrderFactory.makeVarOrders( new VariableOrder(new String[]{"name", "age"}) );
+
+ // Create the PCJ table in Accumulo.
+ pcjs.createPcjTable(accumuloConn, pcjTableName, varOrders, sparql);
+ setupTestUsers(pcjTableName);
+
+ // Add a few results to the PCJ table.
+ final MapBindingSet alice = new MapBindingSet();
+ alice.addBinding("name", new URIImpl("http://Alice"));
+ alice.addBinding("age", new NumericLiteralImpl(14, XMLSchema.INTEGER));
+
+ final MapBindingSet bob = new MapBindingSet();
+ bob.addBinding("name", new URIImpl("http://Bob"));
+ bob.addBinding("age", new NumericLiteralImpl(16, XMLSchema.INTEGER));
+
+ final MapBindingSet charlie = new MapBindingSet();
+ charlie.addBinding("name", new URIImpl("http://Charlie"));
+ charlie.addBinding("age", new NumericLiteralImpl(12, XMLSchema.INTEGER));
+
+ final VisibilityBindingSet aliceVisibility = new VisibilityBindingSet(alice, "A&B&C");
+ final VisibilityBindingSet bobVisibility = new VisibilityBindingSet(bob, "B&C");
+ final VisibilityBindingSet charlieVisibility = new VisibilityBindingSet(charlie, "C");
+
+ final Set<BindingSet> results = Sets.<BindingSet>newHashSet(alice, bob, charlie);
+ final Set<VisibilityBindingSet> visibilityResults = Sets.<VisibilityBindingSet>newHashSet(aliceVisibility, bobVisibility, charlieVisibility);
+ // Load historic matches from Rya into the PCJ table.
+ pcjs.addResults(accumuloConn, pcjTableName, visibilityResults);
+
+ // Make sure the cardinality was updated.
+ final PcjMetadata metadata = pcjs.getPcjMetadata(accumuloConn, pcjTableName);
+ assertEquals(3, metadata.getCardinality());
+
+ // Scan Accumulo for the stored results.
+ Multimap<String, BindingSet> fetchedResults = loadPcjResults(accumuloConn, pcjTableName, "A", "B", "C");
+ assertEquals(getExpectedResults(results), fetchedResults);
+ fetchedResults = loadPcjResults(accumuloConn, pcjTableName, "C", "B");
+ assertEquals(getExpectedResults(Sets.<BindingSet>newHashSet(bob, charlie)), fetchedResults);
+ fetchedResults = loadPcjResults(accumuloConn, pcjTableName, "C");
+ assertEquals(getExpectedResults(Sets.<BindingSet>newHashSet(charlie)), fetchedResults);
+
+ final Connector cConn = accumuloConn.getInstance().getConnector("cUser", new PasswordToken("password"));
+ // Scan Accumulo for the stored results.
+ fetchedResults = loadPcjResults(cConn, pcjTableName, "C");
+ assertEquals(getExpectedResults(Sets.<BindingSet>newHashSet(charlie)), fetchedResults);
+ fetchedResults = loadPcjResults(cConn, pcjTableName);
+ assertEquals(getExpectedResults(Sets.<BindingSet>newHashSet(charlie)), fetchedResults);
+
+ final Connector noAuthConn = accumuloConn.getInstance().getConnector("noAuth", new PasswordToken("password"));
+ // Scan Accumulo for the stored results.
+ fetchedResults = loadPcjResults(noAuthConn, pcjTableName);
+ assertTrue(fetchedResults.isEmpty());
+ }
+
+ @Test
+ public void createWithVisibilityFluo() throws Exception {
+ final String sparql =
+ "SELECT ?customer ?worker ?city " +
+ "{ " +
+ "?customer <http://talksTo> ?worker. " +
+ "?worker <http://livesIn> ?city. " +
+ "?worker <http://worksAt> <http://Chipotle>. " +
+ "}";
+
+ // Triples that will be streamed into Fluo after the PCJ has been created.
+ final Map<RyaStatement, String> streamedTriples = new HashMap<>();
+ addStatementVisibilityEntry(streamedTriples, makeRyaStatement("http://Alice", "http://talksTo", "http://Bob"), "A&B");
+ addStatementVisibilityEntry(streamedTriples, makeRyaStatement("http://Bob", "http://livesIn", "http://London"), "A");
+ addStatementVisibilityEntry(streamedTriples, makeRyaStatement("http://Bob", "http://worksAt", "http://Chipotle"), "B");
+
+ addStatementVisibilityEntry(streamedTriples, makeRyaStatement("http://Alice", "http://talksTo", "http://Charlie"), "B&C");
+ addStatementVisibilityEntry(streamedTriples, makeRyaStatement("http://Charlie", "http://livesIn", "http://London"), "B");
+ addStatementVisibilityEntry(streamedTriples, makeRyaStatement("http://Charlie", "http://worksAt", "http://Chipotle"), "C");
+
+ addStatementVisibilityEntry(streamedTriples, makeRyaStatement("http://Alice", "http://talksTo", "http://David"), "C&D");
+ addStatementVisibilityEntry(streamedTriples, makeRyaStatement("http://David", "http://livesIn", "http://London"), "C");
+ addStatementVisibilityEntry(streamedTriples, makeRyaStatement("http://David", "http://worksAt", "http://Chipotle"), "D");
+
+ addStatementVisibilityEntry(streamedTriples, makeRyaStatement("http://Alice", "http://talksTo", "http://Eve"), "D&E");
+ addStatementVisibilityEntry(streamedTriples, makeRyaStatement("http://Eve", "http://livesIn", "http://Leeds"), "D");
+ addStatementVisibilityEntry(streamedTriples, makeRyaStatement("http://Eve", "http://worksAt", "http://Chipotle"), "E");
+
+ addStatementVisibilityEntry(streamedTriples, makeRyaStatement("http://Frank", "http://talksTo", "http://Alice"), "");
+ addStatementVisibilityEntry(streamedTriples, makeRyaStatement("http://Frank", "http://livesIn", "http://London"), "");
+ addStatementVisibilityEntry(streamedTriples, makeRyaStatement("http://Frank", "http://worksAt", "http://Chipotle"), "");
+
+ // Create the PCJ in Fluo.
+ new CreatePcj().withRyaIntegration(fluoClient, RYA_TABLE_PREFIX, ryaRepo, accumuloConn, new HashSet<VariableOrder>(), sparql);
+
+ // Stream the data into Fluo.
+ for(final RyaStatement statement : streamedTriples.keySet()) {
+ new InsertTriples().insert(fluoClient, statement, Optional.of(streamedTriples.get(statement)));
+ }
+
+ // Fetch the exported results from Accumulo once the observers finish working.
+ fluo.waitForObservers();
+
+ // Fetch expected results from the PCJ table that is in Accumulo.
+ final String exportTableName;
+ try(Snapshot snapshot = fluoClient.newSnapshot()) {
+ final Bytes queryId = snapshot.get(Bytes.of(sparql), FluoQueryColumns.QUERY_ID);
+ exportTableName = snapshot.get(queryId, FluoQueryColumns.QUERY_RYA_EXPORT_TABLE_NAME).toString();
+ }
+ setupTestUsers(exportTableName);
+ Multimap<String, BindingSet> results = loadPcjResults(accumuloConn, exportTableName);
+
+ // Verify the end results of the query match the expected results.
+ Multimap<String, BindingSet> expected = makeExpected(
+ makeBindingSet(
+ new BindingImpl("customer", new URIImpl("http://Alice")),
+ new BindingImpl("worker", new URIImpl("http://Bob")),
+ new BindingImpl("city", new URIImpl("http://London"))),
+ makeBindingSet(
+ new BindingImpl("customer", new URIImpl("http://Alice")),
+ new BindingImpl("worker", new URIImpl("http://Charlie")),
+ new BindingImpl("city", new URIImpl("http://London"))),
+ makeBindingSet(
+ new BindingImpl("customer", new URIImpl("http://Alice")),
+ new BindingImpl("worker", new URIImpl("http://Eve")),
+ new BindingImpl("city", new URIImpl("http://Leeds"))),
+ makeBindingSet(
+ new BindingImpl("customer", new URIImpl("http://Alice")),
+ new BindingImpl("worker", new URIImpl("http://David")),
+ new BindingImpl("city", new URIImpl("http://London"))));
+
+ assertEquals(expected, results);
+
+ final PasswordToken pass = new PasswordToken("password");
+ Connector userConn = accumuloConn.getInstance().getConnector("abUser", pass);
+ results = loadPcjResults(userConn, exportTableName);
+ // Verify the end results of the query match the expected results.
+ expected = makeExpected(
+ makeBindingSet(
+ new BindingImpl("customer", new URIImpl("http://Alice")),
+ new BindingImpl("worker", new URIImpl("http://Bob")),
+ new BindingImpl("city", new URIImpl("http://London"))));
+ assertEquals(expected, results);
+
+ userConn = accumuloConn.getInstance().getConnector("abcUser", pass);
+ results = loadPcjResults(userConn, exportTableName);
+ expected = makeExpected(
+ makeBindingSet(
+ new BindingImpl("customer", new URIImpl("http://Alice")),
+ new BindingImpl("worker", new URIImpl("http://Bob")),
+ new BindingImpl("city", new URIImpl("http://London"))),
+ makeBindingSet(
+ new BindingImpl("customer", new URIImpl("http://Alice")),
+ new BindingImpl("worker", new URIImpl("http://Charlie")),
+ new BindingImpl("city", new URIImpl("http://London"))));
+ assertEquals(expected, results);
+
+ userConn = accumuloConn.getInstance().getConnector("adeUser", pass);
+ results = loadPcjResults(userConn, exportTableName);
+ expected = makeExpected(
+ makeBindingSet(
+ new BindingImpl("customer", new URIImpl("http://Alice")),
+ new BindingImpl("worker", new URIImpl("http://Eve")),
+ new BindingImpl("city", new URIImpl("http://Leeds"))));
+ assertEquals(expected, results);
+
+ userConn = accumuloConn.getInstance().getConnector("noAuth", pass);
+ results = loadPcjResults(userConn, exportTableName);
+ assertTrue(results.isEmpty());
+ }
+
+ private Multimap<String, BindingSet> makeExpected(final BindingSet... bindingSets) {
+ final Set<BindingSet> expectedResults = new HashSet<>();
+ for(final BindingSet bs : bindingSets) {
+ expectedResults.add(bs);
+ }
+
+ final Multimap<String, BindingSet> expected = HashMultimap.create();
+ expected.putAll("customer;worker;city", expectedResults);
+ expected.putAll("worker;city;customer", expectedResults);
+ expected.putAll("city;customer;worker", expectedResults);
+ return expected;
+ }
+
+ private void setupTestUsers(final String exportTableName) throws AccumuloException, AccumuloSecurityException {
+ final PasswordToken pass = new PasswordToken("password");
+ final SecurityOperations secOps = accumuloConn.securityOperations();
+ secOps.changeUserAuthorizations("root", new Authorizations("A", "B", "C", "D", "E"));
+ secOps.createLocalUser("abUser", pass);
+ secOps.changeUserAuthorizations("abUser", new Authorizations("A", "B"));
+ secOps.grantTablePermission("abUser", exportTableName, TablePermission.READ);
+
+ secOps.createLocalUser("abcUser", pass);
+ secOps.changeUserAuthorizations("abcUser", new Authorizations("A", "B", "C"));
+ secOps.grantTablePermission("abcUser", exportTableName, TablePermission.READ);
+
+ secOps.createLocalUser("adeUser", pass);
+ secOps.changeUserAuthorizations("adeUser", new Authorizations("A", "D", "E"));
+ secOps.grantTablePermission("adeUser", exportTableName, TablePermission.READ);
+
+ secOps.createLocalUser("cUser", pass);
+ secOps.changeUserAuthorizations("cUser", new Authorizations("C"));
+ secOps.grantTablePermission("cUser", exportTableName, TablePermission.READ);
+
+ secOps.createLocalUser("noAuth", pass);
+ secOps.changeUserAuthorizations("noAuth", new Authorizations());
+ secOps.grantTablePermission("noAuth", exportTableName, TablePermission.READ);
+ }
+
+
+ private Multimap<String, BindingSet> getExpectedResults(final Set<BindingSet> results) {
+ final Multimap<String, BindingSet> expectedResults = HashMultimap.create();
+ expectedResults.putAll("name;age", results);
+ expectedResults.putAll("age;name", results);
+ return expectedResults;
+ }
+
+ protected static void addStatementVisibilityEntry(final Map<RyaStatement, String> triplesMap, final RyaStatement statement, final String visibility) {
+ triplesMap.put(statement, visibility);
+ }
+
+ /**
+ * Scan accumulo for the results that are stored in a PCJ table. The
+ * multimap stores a set of deserialized binding sets that were in the PCJ
+ * table for every variable order that is found in the PCJ metadata.
+ * @throws AccumuloSecurityException
+ * @throws AccumuloException
+ */
+ private static Multimap<String, BindingSet> loadPcjResults(final Connector accumuloConn, final String pcjTableName, final String... visibility) throws PcjException, TableNotFoundException, BindingSetConversionException, AccumuloException, AccumuloSecurityException {
+ final Multimap<String, BindingSet> fetchedResults = HashMultimap.create();
+
+ final Authorizations userAuths;
+ if(visibility.length == 0) {
+ userAuths = accumuloConn.securityOperations().getUserAuthorizations(accumuloConn.whoami());
+ } else {
+ userAuths = new Authorizations(visibility);
+ }
+
+ // Get the variable orders the data was written to.
+ final PcjTables pcjs = new PcjTables();
+ final PcjMetadata pcjMetadata = pcjs.getPcjMetadata(accumuloConn, pcjTableName);
+
+ // Scan Accumulo for the stored results.
+ for(final VariableOrder varOrder : pcjMetadata.getVarOrders()) {
+ final Scanner scanner = accumuloConn.createScanner(pcjTableName, userAuths);
+ scanner.fetchColumnFamily( new Text(varOrder.toString()) );
+
+ for(final Entry<Key, Value> entry : scanner) {
+ final byte[] serializedResult = entry.getKey().getRow().getBytes();
+ final BindingSet result = converter.convert(serializedResult, varOrder);
+ fetchedResults.put(varOrder.toString(), result);
+ }
+ }
+
+ return fetchedResults;
+ }
+}
[3/3] incubator-rya git commit: RYA-53 Added visibility support
Posted by pu...@apache.org.
RYA-53 Added visibility support
Visibility support added for PCJ tables.
If no visibilities are defined during the query,
only binding sets visible to the accesssor will
be returned.
Added and updated tests for PCJ visibility.
Added visibility support for Fluo backed Rya
Added integration tests for inserting
triples through fluo into PCJs with
visibility.
Added a Decorator pattern for BindingSets
Project: http://git-wip-us.apache.org/repos/asf/incubator-rya/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-rya/commit/c53b5402
Tree: http://git-wip-us.apache.org/repos/asf/incubator-rya/tree/c53b5402
Diff: http://git-wip-us.apache.org/repos/asf/incubator-rya/diff/c53b5402
Branch: refs/heads/develop
Commit: c53b5402b522c1d4769a5f2553c6ffcfe2c8b2c0
Parents: 8168b85
Author: isper3at <sm...@gmail.com>
Authored: Mon Mar 21 12:55:56 2016 -0400
Committer: pujav65 <pu...@gmail.com>
Committed: Fri May 13 09:00:50 2016 -0400
----------------------------------------------------------------------
.../indexing/accumulo/VisibilityBindingSet.java | 90 +++++
.../indexing/external/BindingSetDecorator.java | 105 ++++++
.../external/tupleSet/BindingSetConverter.java | 4 +-
.../tupleSet/BindingSetStringConverter.java | 28 +-
.../indexing/external/tupleSet/PcjTables.java | 110 +++---
.../VisibilityBindingSetStringConverter.java | 62 ++++
.../tupleSet/BindingSetStringConverterTest.java | 54 +--
.../tupleSet/PcjTablesIntegrationTests.java | 123 ++++---
...VisibilityBindingSetStringConverterTest.java | 132 +++++++
.../rya/indexing/pcj/fluo/api/CreatePcj.java | 10 +-
.../indexing/pcj/fluo/api/InsertTriples.java | 18 +-
.../pcj/fluo/app/FilterResultUpdater.java | 18 +-
.../rya/indexing/pcj/fluo/app/IncUpdateDAO.java | 2 +-
.../pcj/fluo/app/JoinResultUpdater.java | 90 +++--
.../pcj/fluo/app/QueryResultUpdater.java | 10 +-
.../app/export/IncrementalResultExporter.java | 5 +-
.../fluo/app/export/rya/RyaResultExporter.java | 4 +-
.../fluo/app/observers/BindingSetUpdater.java | 18 +-
.../pcj/fluo/app/observers/FilterObserver.java | 7 +-
.../pcj/fluo/app/observers/JoinObserver.java | 7 +-
.../fluo/app/observers/QueryResultObserver.java | 20 +-
.../app/observers/StatementPatternObserver.java | 8 +-
.../pcj/fluo/app/observers/TripleObserver.java | 42 ++-
.../pcj/fluo/app/query/FluoQueryColumns.java | 2 +-
.../fluo/app/query/FluoQueryMetadataDAO.java | 4 +-
.../pcj/fluo/app/LeftOuterJoinTest.java | 82 +++--
.../indexing/pcj/fluo/app/NaturalJoinTest.java | 70 ++--
.../pcj/fluo/client/util/FluoLoader.java | 6 +-
.../pcj/fluo/demo/FluoAndHistoricPcjsDemo.java | 6 +-
.../pcj/fluo/api/CountStatementsIT.java | 3 +-
.../indexing/pcj/fluo/api/GetQueryReportIT.java | 3 +-
.../indexing/pcj/fluo/integration/InputIT.java | 7 +-
.../indexing/pcj/fluo/integration/QueryIT.java | 9 +-
.../pcj/fluo/integration/RyaExportIT.java | 6 +-
.../pcj/fluo/visibility/PcjVisibilityIT.java | 362 +++++++++++++++++++
35 files changed, 1183 insertions(+), 344 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/indexing/src/main/java/mvm/rya/indexing/accumulo/VisibilityBindingSet.java
----------------------------------------------------------------------
diff --git a/extras/indexing/src/main/java/mvm/rya/indexing/accumulo/VisibilityBindingSet.java b/extras/indexing/src/main/java/mvm/rya/indexing/accumulo/VisibilityBindingSet.java
new file mode 100644
index 0000000..b9e2351
--- /dev/null
+++ b/extras/indexing/src/main/java/mvm/rya/indexing/accumulo/VisibilityBindingSet.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package mvm.rya.indexing.accumulo;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import javax.annotation.ParametersAreNonnullByDefault;
+
+import org.openrdf.query.BindingSet;
+
+import mvm.rya.indexing.external.BindingSetDecorator;
+
+/**
+ * Decorates a {@link BindingSet} with a collection of visibilities.
+ */
+@ParametersAreNonnullByDefault
+public class VisibilityBindingSet extends BindingSetDecorator {
+ private static final long serialVersionUID = 1L;
+ private final String visibility;
+ private volatile int hashCode;
+
+ /**
+ * @param set - Decorates the {@link BindingSet} with no visibilities.
+ */
+ public VisibilityBindingSet(final BindingSet set) {
+ this(set, "");
+ }
+
+ /**
+ * Creates a new {@link VisibilityBindingSet}
+ * @param set - The {@link BindingSet} to decorate
+ * @param visibility - The visibilities on the {@link BindingSet} (not null)
+ */
+ public VisibilityBindingSet(final BindingSet set, final String visibility) {
+ super(set);
+ this.visibility = checkNotNull(visibility);
+ }
+
+ /**
+ * @return - The Visibilities on the {@link BindingSet}
+ */
+ public String getVisibility() {
+ return visibility;
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) {
+ return true;
+ } else if(o instanceof VisibilityBindingSet) {
+ final VisibilityBindingSet other = (VisibilityBindingSet) o;
+ return set.equals(other) && visibility.equals(other.getVisibility());
+ }
+ return false;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = hashCode;
+ if(result == 0) {
+ result = 31 * result + visibility.hashCode();
+ result = 31 * result + super.hashCode();
+ hashCode = result;
+ }
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ final StringBuilder sb = new StringBuilder(super.toString());
+ sb.append("\n Visibility: " + getVisibility() + "\n");
+ return sb.toString();
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/indexing/src/main/java/mvm/rya/indexing/external/BindingSetDecorator.java
----------------------------------------------------------------------
diff --git a/extras/indexing/src/main/java/mvm/rya/indexing/external/BindingSetDecorator.java b/extras/indexing/src/main/java/mvm/rya/indexing/external/BindingSetDecorator.java
new file mode 100644
index 0000000..b4909bd
--- /dev/null
+++ b/extras/indexing/src/main/java/mvm/rya/indexing/external/BindingSetDecorator.java
@@ -0,0 +1,105 @@
+/*
+ * 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 mvm.rya.indexing.external;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.Iterator;
+import java.util.Set;
+
+import org.openrdf.model.Value;
+import org.openrdf.query.Binding;
+import org.openrdf.query.BindingSet;
+
+/**
+ * Abstracts out the decoration of a {@link BindingSet}.
+ */
+public abstract class BindingSetDecorator implements BindingSet {
+ private static final long serialVersionUID = 1L;
+ protected final BindingSet set;
+ private volatile int hashCode;
+
+ /**
+ * Constructs a new {@link BindingSetDecorator}, decorating the provided
+ * {@link BindingSet}.
+ * @param set - The {@link BindingSet} to be decorated. (not null)
+ */
+ public BindingSetDecorator(final BindingSet set) {
+ this.set = checkNotNull(set);
+ }
+
+ @Override
+ public Iterator<Binding> iterator() {
+ return set.iterator();
+ }
+
+ @Override
+ public Set<String> getBindingNames() {
+ return set.getBindingNames();
+ }
+
+ @Override
+ public Binding getBinding(final String bindingName) {
+ return set.getBinding(bindingName);
+ }
+
+ @Override
+ public boolean hasBinding(final String bindingName) {
+ return set.hasBinding(bindingName);
+ }
+
+ @Override
+ public Value getValue(final String bindingName) {
+ return set.getValue(bindingName);
+ }
+
+ @Override
+ public int size() {
+ return set.size();
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if(!(o instanceof BindingSetDecorator)) {
+ return false;
+ }
+ final BindingSetDecorator other = (BindingSetDecorator) o;
+ return set.equals(other.set);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = hashCode;
+ if(result == 0) {
+ result = 31 * result + set.hashCode();
+ hashCode = result;
+ }
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ final StringBuilder sb = new StringBuilder();
+ sb.append(" names: ");
+ for (final String name : getBindingNames()) {
+ sb.append("\n [name]: " + name + " --- [value]: " + getBinding(name).getValue().toString());
+ }
+ return sb.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/BindingSetConverter.java
----------------------------------------------------------------------
diff --git a/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/BindingSetConverter.java b/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/BindingSetConverter.java
index 9c10dfa..84aee0c 100644
--- a/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/BindingSetConverter.java
+++ b/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/BindingSetConverter.java
@@ -91,7 +91,7 @@ public interface BindingSetConverter<T> {
*
* @param message - Describes why this exception was thrown.
*/
- public BindingSetConversionException(String message) {
+ public BindingSetConversionException(final String message) {
super(message);
}
@@ -101,7 +101,7 @@ public interface BindingSetConverter<T> {
* @param message - Describes why this exception was thrown.
* @param cause - The exception that caused this one to be thrown.
*/
- public BindingSetConversionException(String message, Throwable cause) {
+ public BindingSetConversionException(final String message, final Throwable cause) {
super(message, cause);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/BindingSetStringConverter.java
----------------------------------------------------------------------
diff --git a/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/BindingSetStringConverter.java b/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/BindingSetStringConverter.java
index 6d69d5b..c4116f2 100644
--- a/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/BindingSetStringConverter.java
+++ b/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/BindingSetStringConverter.java
@@ -57,24 +57,24 @@ public class BindingSetStringConverter implements BindingSetConverter<String> {
private static final ValueFactory valueFactory = new ValueFactoryImpl();
@Override
- public String convert(BindingSet bindingSet, VariableOrder varOrder) {
+ public String convert(final BindingSet bindingSet, final VariableOrder varOrder) {
checkBindingsSubsetOfVarOrder(bindingSet, varOrder);
// Convert each Binding to a String.
- List<String> bindingStrings = new ArrayList<>();
- for(String varName : varOrder) {
+ final List<String> bindingStrings = new ArrayList<>();
+ for(final String varName : varOrder) {
if(bindingSet.hasBinding(varName)) {
// Add a value to the binding set.
final Value value = bindingSet.getBinding(varName).getValue();
final RyaType ryaValue = RdfToRyaConversions.convertValue(value);
- String bindingString = ryaValue.getData() + TYPE_DELIM + ryaValue.getDataType();
+ final String bindingString = ryaValue.getData() + TYPE_DELIM + ryaValue.getDataType();
bindingStrings.add(bindingString);
} else {
// Add a null value to the binding set.
- bindingStrings.add(NULL_VALUE_STRING);
+ bindingStrings.add(NULL_VALUE_STRING);
}
}
-
+
// Join the bindings using the binding delim.
return Joiner.on(BINDING_DELIM).join(bindingStrings);
}
@@ -82,23 +82,23 @@ public class BindingSetStringConverter implements BindingSetConverter<String> {
/**
* Checks to see if the names of all the {@link Binding}s in the {@link BindingSet}
* are a subset of the variables names in {@link VariableOrder}.
- *
+ *
* @param bindingSet - The binding set whose Bindings will be inspected. (not null)
* @param varOrder - The names of the bindings that may appear in the BindingSet. (not null)
* @throws IllegalArgumentException Indicates the names of the bindings are
- * not a subset of the variable order.
+ * not a subset of the variable order.
*/
- private static void checkBindingsSubsetOfVarOrder(BindingSet bindingSet, VariableOrder varOrder) throws IllegalArgumentException {
+ private static void checkBindingsSubsetOfVarOrder(final BindingSet bindingSet, final VariableOrder varOrder) throws IllegalArgumentException {
checkNotNull(bindingSet);
checkNotNull(varOrder);
- Set<String> bindingNames = bindingSet.getBindingNames();
- List<String> varNames = varOrder.getVariableOrders();
+ final Set<String> bindingNames = bindingSet.getBindingNames();
+ final List<String> varNames = varOrder.getVariableOrders();
checkArgument(varNames.containsAll(bindingNames), "The BindingSet contains a Binding whose name is not part of the VariableOrder.");
}
@Override
- public BindingSet convert(String bindingSetString, VariableOrder varOrder) {
+ public BindingSet convert(final String bindingSetString, final VariableOrder varOrder) {
checkNotNull(bindingSetString);
checkNotNull(varOrder);
@@ -108,7 +108,7 @@ public class BindingSetStringConverter implements BindingSetConverter<String> {
final QueryBindingSet bindingSet = new QueryBindingSet();
for(int i = 0; i < bindingStrings.length; i++) {
- String bindingString = bindingStrings[i];
+ final String bindingString = bindingStrings[i];
if(!NULL_VALUE_STRING.equals(bindingString)) {
final String name = varOrrderArr[i];
final Value value = toValue(bindingStrings[i]);
@@ -124,7 +124,7 @@ public class BindingSetStringConverter implements BindingSetConverter<String> {
* @param valueString - The String representation of the value. (not null)
* @return The {@link Value} representation of the String.
*/
- private static Value toValue(final String valueString) {
+ protected static Value toValue(final String valueString) {
checkNotNull(valueString);
// Split the String that was stored in Fluo into its Value and Type parts.
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/PcjTables.java
----------------------------------------------------------------------
diff --git a/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/PcjTables.java b/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/PcjTables.java
index fc940f6..d8a81a6 100644
--- a/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/PcjTables.java
+++ b/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/PcjTables.java
@@ -56,6 +56,7 @@ import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Mutation;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.security.ColumnVisibility;
import org.apache.commons.lang3.builder.EqualsBuilder;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Logger;
@@ -74,7 +75,7 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
-import mvm.rya.api.resolver.RyaTypeResolverException;
+import mvm.rya.indexing.accumulo.VisibilityBindingSet;
import mvm.rya.indexing.external.tupleSet.BindingSetConverter.BindingSetConversionException;
/**
@@ -132,9 +133,9 @@ public class PcjTables {
*
* @param varOrder - An ordered array of Binding Set variables. (not null)
*/
- public VariableOrder(String... varOrder) {
+ public VariableOrder(final String... varOrder) {
checkNotNull(varOrder);
- this.variableOrder = ImmutableList.copyOf(varOrder);
+ variableOrder = ImmutableList.copyOf(varOrder);
}
/**
@@ -142,9 +143,9 @@ public class PcjTables {
*
* @param varOrder - An ordered collection of Binding Set variables. (not null)
*/
- public VariableOrder(Collection<String> varOrder) {
+ public VariableOrder(final Collection<String> varOrder) {
checkNotNull(varOrder);
- this.variableOrder = ImmutableList.copyOf(varOrder);
+ variableOrder = ImmutableList.copyOf(varOrder);
}
/**
@@ -152,9 +153,9 @@ public class PcjTables {
*
* @param varOrderString - The String representation of a VariableOrder. (not null)
*/
- public VariableOrder(String varOrderString) {
+ public VariableOrder(final String varOrderString) {
checkNotNull(varOrderString);
- this.variableOrder = ImmutableList.copyOf( varOrderString.split(VAR_ORDER_DELIM) );
+ variableOrder = ImmutableList.copyOf( varOrderString.split(VAR_ORDER_DELIM) );
}
/**
@@ -168,7 +169,7 @@ public class PcjTables {
* @return The variable order as an ordered array of Strings. This array is mutable.
*/
public String[] toArray() {
- String[] array = new String[ variableOrder.size() ];
+ final String[] array = new String[ variableOrder.size() ];
return variableOrder.toArray( array );
}
@@ -183,11 +184,11 @@ public class PcjTables {
}
@Override
- public boolean equals(Object o) {
+ public boolean equals(final Object o) {
if(this == o) {
return true;
} else if(o instanceof VariableOrder) {
- VariableOrder varOrder = (VariableOrder) o;
+ final VariableOrder varOrder = (VariableOrder) o;
return variableOrder.equals( varOrder.variableOrder );
}
return false;
@@ -342,7 +343,7 @@ public class PcjTables {
public static class ShiftVarOrderFactory implements PcjVarOrderFactory {
@Override
public Set<VariableOrder> makeVarOrders(final VariableOrder varOrder) {
- Set<VariableOrder> varOrders = new HashSet<>();
+ final Set<VariableOrder> varOrders = new HashSet<>();
final List<String> cyclicBuff = Lists.newArrayList( varOrder.getVariableOrders() );
final String[] varOrderBuff = new String[ cyclicBuff.size() ];
@@ -373,7 +374,7 @@ public class PcjTables {
*
* @param message - Describes why the exception is being thrown.
*/
- public PcjException(String message) {
+ public PcjException(final String message) {
super(message);
}
@@ -383,7 +384,7 @@ public class PcjTables {
* @param message - Describes why the exception is being thrown.
* @param cause - The exception that caused this one to be thrown.
*/
- public PcjException(String message, Throwable cause) {
+ public PcjException(final String message, final Throwable cause) {
super(message, cause);
}
}
@@ -470,24 +471,24 @@ public class PcjTables {
// SPARQL Query
Mutation mutation = new Mutation(PCJ_METADATA_ROW_ID);
- Value query = new Value( stringLexicoder.encode(metadata.getSparql()) );
+ final Value query = new Value( stringLexicoder.encode(metadata.getSparql()) );
mutation.put(PCJ_METADATA_FAMILY, PCJ_METADATA_SPARQL_QUERY, query);
mutations.add(mutation);
// Cardinality
mutation = new Mutation(PCJ_METADATA_ROW_ID);
- Value cardinality = new Value( longLexicoder.encode(new Long(metadata.getCardinality())) );
+ final Value cardinality = new Value( longLexicoder.encode(new Long(metadata.getCardinality())) );
mutation.put(PCJ_METADATA_FAMILY, PCJ_METADATA_CARDINALITY, cardinality);
mutations.add(mutation);
// Variable Orders
- List<String> varOrderStrings = new ArrayList<>();
- for(VariableOrder varOrder : metadata.getVarOrders()) {
+ final List<String> varOrderStrings = new ArrayList<>();
+ for(final VariableOrder varOrder : metadata.getVarOrders()) {
varOrderStrings.add( varOrder.toString() );
}
mutation = new Mutation(PCJ_METADATA_ROW_ID);
- Value variableOrders = new Value( listLexicoder.encode(varOrderStrings) );
+ final Value variableOrders = new Value( listLexicoder.encode(varOrderStrings) );
mutation.put(PCJ_METADATA_FAMILY, PCJ_METADATA_VARIABLE_ORDERS, variableOrders);
mutations.add(mutation);
@@ -512,7 +513,7 @@ public class PcjTables {
try {
// Create an Accumulo scanner that iterates through the metadata entries.
- Scanner scanner = accumuloConn.createScanner(pcjTableName, new Authorizations());
+ final Scanner scanner = accumuloConn.createScanner(pcjTableName, new Authorizations());
final Iterator<Entry<Key, Value>> entries = scanner.iterator();
// No metadata has been stored in the table yet.
@@ -527,15 +528,15 @@ public class PcjTables {
while(entries.hasNext()) {
final Entry<Key, Value> entry = entries.next();
- Text columnQualifier = entry.getKey().getColumnQualifier();
- byte[] value = entry.getValue().get();
+ final Text columnQualifier = entry.getKey().getColumnQualifier();
+ final byte[] value = entry.getValue().get();
if(columnQualifier.equals(PCJ_METADATA_SPARQL_QUERY)) {
sparql = stringLexicoder.decode(value);
} else if(columnQualifier.equals(PCJ_METADATA_CARDINALITY)) {
cardinality = longLexicoder.decode(value);
} else if(columnQualifier.equals(PCJ_METADATA_VARIABLE_ORDERS)) {
- for(String varOrderStr : listLexicoder.decode(value)) {
+ for(final String varOrderStr : listLexicoder.decode(value)) {
varOrders.add( new VariableOrder(varOrderStr) );
}
}
@@ -543,7 +544,7 @@ public class PcjTables {
return new PcjMetadata(sparql, cardinality, varOrders);
- } catch (TableNotFoundException e) {
+ } catch (final TableNotFoundException e) {
throw new PcjException("Could not add results to a PCJ because the PCJ table does not exist.", e);
}
}
@@ -563,7 +564,7 @@ public class PcjTables {
public void addResults(
final Connector accumuloConn,
final String pcjTableName,
- final Collection<BindingSet> results) throws PcjException {
+ final Collection<VisibilityBindingSet> results) throws PcjException {
checkNotNull(accumuloConn);
checkNotNull(pcjTableName);
checkNotNull(results);
@@ -587,20 +588,20 @@ public class PcjTables {
private void writeResults(
final Connector accumuloConn,
final String pcjTableName,
- final Collection<BindingSet> results) throws PcjException {
+ final Collection<VisibilityBindingSet> results) throws PcjException {
checkNotNull(accumuloConn);
checkNotNull(pcjTableName);
checkNotNull(results);
// Fetch the variable orders from the PCJ table.
- PcjMetadata metadata = getPcjMetadata(accumuloConn, pcjTableName);
+ final PcjMetadata metadata = getPcjMetadata(accumuloConn, pcjTableName);
// Write each result formatted using each of the variable orders.
BatchWriter writer = null;
try {
writer = accumuloConn.createBatchWriter(pcjTableName, new BatchWriterConfig());
- for(BindingSet result : results) {
- Set<Mutation> addResultMutations = makeWriteResultMutations(metadata.getVarOrders(), result);
+ for(final VisibilityBindingSet result : results) {
+ final Set<Mutation> addResultMutations = makeWriteResultMutations(metadata.getVarOrders(), result);
writer.addMutations( addResultMutations );
}
} catch (TableNotFoundException | MutationsRejectedException e) {
@@ -609,7 +610,7 @@ public class PcjTables {
if(writer != null) {
try {
writer.close();
- } catch (MutationsRejectedException e) {
+ } catch (final MutationsRejectedException e) {
throw new PcjException("Could not add results to a PCJ table because some of the mutations were rejected.", e);
}
}
@@ -627,23 +628,24 @@ public class PcjTables {
*/
private static Set<Mutation> makeWriteResultMutations(
final Set<VariableOrder> varOrders,
- final BindingSet result) throws PcjException {
+ final VisibilityBindingSet result) throws PcjException {
checkNotNull(varOrders);
checkNotNull(result);
- Set<Mutation> mutations = new HashSet<>();
- AccumuloPcjSerializer converter = new AccumuloPcjSerializer();
-
+ final Set<Mutation> mutations = new HashSet<>();
+ final AccumuloPcjSerializer converter = new AccumuloPcjSerializer();
+
for(final VariableOrder varOrder : varOrders) {
try {
// Serialize the result to the variable order.
- byte[] serializedResult = converter.convert(result, varOrder);
+ final byte[] serializedResult = converter.convert(result, varOrder);
// Row ID = binding set values, Column Family = variable order of the binding set.
- Mutation addResult = new Mutation(serializedResult);
- addResult.put(varOrder.toString(), "", "");
+ final Mutation addResult = new Mutation(serializedResult);
+ final String visibility = result.getVisibility();
+ addResult.put(varOrder.toString(), "", new ColumnVisibility(visibility), "");
mutations.add(addResult);
- } catch(BindingSetConversionException e) {
+ } catch(final BindingSetConversionException e) {
throw new PcjException("Could not serialize a result.", e);
}
}
@@ -673,9 +675,9 @@ public class PcjTables {
boolean updated = false;
while(!updated) {
// Write the conditional update request to Accumulo.
- long cardinality = getPcjMetadata(accumuloConn, pcjTableName).getCardinality();
- ConditionalMutation mutation = makeUpdateCardinalityMutation(cardinality, delta);
- ConditionalWriter.Result result = conditionalWriter.write(mutation);
+ final long cardinality = getPcjMetadata(accumuloConn, pcjTableName).getCardinality();
+ final ConditionalMutation mutation = makeUpdateCardinalityMutation(cardinality, delta);
+ final ConditionalWriter.Result result = conditionalWriter.write(mutation);
// Interpret the result.
switch(result.getStatus()) {
@@ -688,7 +690,7 @@ public class PcjTables {
// We do not know if the mutation succeeded. At best, we can hope the metadata hasn't been updated
// since we originally fetched it and try again. Otherwise, continue forwards as if it worked. It's
// okay if this number is slightly off.
- long newCardinality = getPcjMetadata(accumuloConn, pcjTableName).getCardinality();
+ final long newCardinality = getPcjMetadata(accumuloConn, pcjTableName).getCardinality();
if(newCardinality != cardinality) {
updated = true;
}
@@ -717,20 +719,20 @@ public class PcjTables {
* @param delta - How much the cardinality will change.
* @return The mutation that will perform the conditional update.
*/
- private static ConditionalMutation makeUpdateCardinalityMutation(long current, long delta) {
+ private static ConditionalMutation makeUpdateCardinalityMutation(final long current, final long delta) {
// Try to update the cardinality by the delta.
- ConditionalMutation mutation = new ConditionalMutation(PCJ_METADATA_ROW_ID);
- Condition lastCardinalityStillCurrent = new Condition(
+ final ConditionalMutation mutation = new ConditionalMutation(PCJ_METADATA_ROW_ID);
+ final Condition lastCardinalityStillCurrent = new Condition(
PCJ_METADATA_FAMILY,
PCJ_METADATA_CARDINALITY);
// Require the old cardinality to be the value we just read.
- byte[] currentCardinalityBytes = longLexicoder.encode( current );
+ final byte[] currentCardinalityBytes = longLexicoder.encode( current );
lastCardinalityStillCurrent.setValue( currentCardinalityBytes );
mutation.addCondition(lastCardinalityStillCurrent);
// If that is the case, then update to the new value.
- Value newCardinality = new Value( longLexicoder.encode(current + delta) );
+ final Value newCardinality = new Value( longLexicoder.encode(current + delta) );
mutation.put(PCJ_METADATA_FAMILY, PCJ_METADATA_CARDINALITY, newCardinality);
return mutation;
}
@@ -757,17 +759,17 @@ public class PcjTables {
try {
// Fetch the query that needs to be executed from the PCJ table.
- PcjMetadata pcjMetadata = getPcjMetadata(accumuloConn, pcjTableName);
- String sparql = pcjMetadata.getSparql();
+ final PcjMetadata pcjMetadata = getPcjMetadata(accumuloConn, pcjTableName);
+ final String sparql = pcjMetadata.getSparql();
// Query Rya for results to the SPARQL query.
- TupleQuery query = ryaConn.prepareTupleQuery(QueryLanguage.SPARQL, sparql);
- TupleQueryResult results = query.evaluate();
+ final TupleQuery query = ryaConn.prepareTupleQuery(QueryLanguage.SPARQL, sparql);
+ final TupleQueryResult results = query.evaluate();
// Load batches of 1000 of them at a time into the PCJ table
- Set<BindingSet> batch = new HashSet<>(1000);
+ final Set<VisibilityBindingSet> batch = new HashSet<>(1000);
while(results.hasNext()) {
- batch.add( results.next() );
+ batch.add( new VisibilityBindingSet(results.next()) );
if(batch.size() == 1000) {
addResults(accumuloConn, pcjTableName, batch);
@@ -819,8 +821,8 @@ public class PcjTables {
checkNotNull(pcjVarOrderFactory);
// Create the PCJ's variable orders.
- PcjVarOrderFactory varOrderFactory = pcjVarOrderFactory.or(DEFAULT_VAR_ORDER_FACTORY);
- Set<VariableOrder> varOrders = varOrderFactory.makeVarOrders( new VariableOrder(resultVariables) );
+ final PcjVarOrderFactory varOrderFactory = pcjVarOrderFactory.or(DEFAULT_VAR_ORDER_FACTORY);
+ final Set<VariableOrder> varOrders = varOrderFactory.makeVarOrders( new VariableOrder(resultVariables) );
// Create the PCJ table in Accumulo.
createPcjTable(accumuloConn, pcjTableName, varOrders, sparql);
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/VisibilityBindingSetStringConverter.java
----------------------------------------------------------------------
diff --git a/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/VisibilityBindingSetStringConverter.java b/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/VisibilityBindingSetStringConverter.java
new file mode 100644
index 0000000..832186c
--- /dev/null
+++ b/extras/indexing/src/main/java/mvm/rya/indexing/external/tupleSet/VisibilityBindingSetStringConverter.java
@@ -0,0 +1,62 @@
+/*
+ * 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 mvm.rya.indexing.external.tupleSet;
+
+import javax.annotation.ParametersAreNonnullByDefault;
+
+import org.openrdf.query.BindingSet;
+
+import com.google.common.base.Strings;
+
+import mvm.rya.indexing.accumulo.VisibilityBindingSet;
+import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
+
+/**
+ * Converts {@link BindingSet}s to Strings and back again. The Strings do not
+ * include the binding names and are ordered with a {@link VariableOrder}.
+ */
+@ParametersAreNonnullByDefault
+public class VisibilityBindingSetStringConverter extends BindingSetStringConverter {
+ public static final char VISIBILITY_DELIM = 1;
+
+ @Override
+ public String convert(final BindingSet bindingSet, final VariableOrder varOrder) {
+ String visibility = "";
+ if(bindingSet instanceof VisibilityBindingSet) {
+ final VisibilityBindingSet visiSet = (VisibilityBindingSet) bindingSet;
+ if(!Strings.isNullOrEmpty(visiSet.getVisibility())) {
+ visibility = VISIBILITY_DELIM + visiSet.getVisibility();
+ }
+ }
+ return super.convert(bindingSet, varOrder) + visibility;
+ }
+
+ @Override
+ public BindingSet convert(final String bindingSetString, final VariableOrder varOrder) {
+ final String[] visiStrings = bindingSetString.split("" + VISIBILITY_DELIM);
+ BindingSet bindingSet = super.convert(visiStrings[0], varOrder);
+
+ if(visiStrings.length > 1) {
+ bindingSet = new VisibilityBindingSet(bindingSet, visiStrings[1]);
+ } else {
+ bindingSet = new VisibilityBindingSet(bindingSet);
+ }
+ return bindingSet;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/indexing/src/test/java/mvm/rya/indexing/external/tupleSet/BindingSetStringConverterTest.java
----------------------------------------------------------------------
diff --git a/extras/indexing/src/test/java/mvm/rya/indexing/external/tupleSet/BindingSetStringConverterTest.java b/extras/indexing/src/test/java/mvm/rya/indexing/external/tupleSet/BindingSetStringConverterTest.java
index 725e557..dc24de0 100644
--- a/extras/indexing/src/test/java/mvm/rya/indexing/external/tupleSet/BindingSetStringConverterTest.java
+++ b/extras/indexing/src/test/java/mvm/rya/indexing/external/tupleSet/BindingSetStringConverterTest.java
@@ -49,7 +49,7 @@ public class BindingSetStringConverterTest {
// Convert it to a String.
final VariableOrder varOrder = new VariableOrder("y", "z", "x");
- BindingSetConverter<String> converter = new BindingSetStringConverter();
+ final BindingSetConverter<String> converter = new BindingSetStringConverter();
final String bindingSetString = converter.convert(originalBindingSet, varOrder);
// Ensure it converted to the expected result.l
@@ -69,7 +69,7 @@ public class BindingSetStringConverterTest {
// Convert it to a String.
final VariableOrder varOrder = new VariableOrder("x");
- BindingSetConverter<String> converter = new BindingSetStringConverter();
+ final BindingSetConverter<String> converter = new BindingSetStringConverter();
final String bindingSetString = converter.convert(originalBindingSet, varOrder);
// Ensure it converted to the expected result.
@@ -85,7 +85,7 @@ public class BindingSetStringConverterTest {
// Convert it to a String.
final VariableOrder varOrder = new VariableOrder("x");
- BindingSetConverter<String> converter = new BindingSetStringConverter();
+ final BindingSetConverter<String> converter = new BindingSetStringConverter();
final String bindingSetString = converter.convert(originalBindingSet, varOrder);
// Ensure it converted to the expected result.
@@ -101,7 +101,7 @@ public class BindingSetStringConverterTest {
// Convert it to a String.
final VariableOrder varOrder = new VariableOrder("x");
- BindingSetConverter<String> converter = new BindingSetStringConverter();
+ final BindingSetConverter<String> converter = new BindingSetStringConverter();
final String bindingSetString = converter.convert(originalBindingSet, varOrder);
// Ensure it converted to the expected result.
@@ -124,11 +124,11 @@ public class BindingSetStringConverterTest {
final VariableOrder varOrder = new VariableOrder("x", "y");
// Create the String representation of the BindingSet.
- BindingSetConverter<String> converter = new BindingSetStringConverter();
- String bindingSetString = converter.convert(originalBindingSet, varOrder);
+ final BindingSetConverter<String> converter = new BindingSetStringConverter();
+ final String bindingSetString = converter.convert(originalBindingSet, varOrder);
// Ensure the expected value was created.
- String expected =
+ final String expected =
"http://a<<~>>http://www.w3.org/2001/XMLSchema#anyURI:::" +
"http://b<<~>>http://www.w3.org/2001/XMLSchema#anyURI";
assertEquals(expected, bindingSetString);
@@ -150,11 +150,11 @@ public class BindingSetStringConverterTest {
final VariableOrder varOrder = new VariableOrder("x", "a", "y", "b");
// Create the String representation of the BindingSet.
- BindingSetConverter<String> converter = new BindingSetStringConverter();
- String bindingSetString = converter.convert(originalBindingSet, varOrder);
+ final BindingSetConverter<String> converter = new BindingSetStringConverter();
+ final String bindingSetString = converter.convert(originalBindingSet, varOrder);
// Ensure the expected value was created.
- String expected =
+ final String expected =
"http://a<<~>>http://www.w3.org/2001/XMLSchema#anyURI:::" +
BindingSetStringConverter.NULL_VALUE_STRING + ":::" +
"http://b<<~>>http://www.w3.org/2001/XMLSchema#anyURI:::" +
@@ -178,7 +178,7 @@ public class BindingSetStringConverterTest {
final VariableOrder varOrder = new VariableOrder("x", "y");
// Create the String representation of the BindingSet. This will throw an exception.
- BindingSetConverter<String> converter = new BindingSetStringConverter();
+ final BindingSetConverter<String> converter = new BindingSetStringConverter();
converter.convert(originalBindingSet, varOrder);
}
@@ -192,7 +192,7 @@ public class BindingSetStringConverterTest {
// Convert it to a BindingSet
final VariableOrder varOrder = new VariableOrder("y", "z", "x");
- BindingSetConverter<String> converter = new BindingSetStringConverter();
+ final BindingSetConverter<String> converter = new BindingSetStringConverter();
final BindingSet bindingSet = converter.convert(bindingSetString, varOrder);
// Ensure it converted to the expected result.
@@ -205,7 +205,7 @@ public class BindingSetStringConverterTest {
}
/**
- * Ensures that when a binding set is converted from a String back to a
+ * Ensures that when a binding set is converted from a String back to a
* BindingSet, null values do not get converted into Bindings.
*/
@Test
@@ -216,27 +216,27 @@ public class BindingSetStringConverterTest {
BindingSetStringConverter.NULL_VALUE_STRING + ":::" +
"http://value 2<<~>>http://www.w3.org/2001/XMLSchema#anyURI:::" +
BindingSetStringConverter.NULL_VALUE_STRING;
-
+
// Convert it to a BindingSet
- VariableOrder varOrder = new VariableOrder("x", "a", "y", "b");
- BindingSetConverter<String> converter = new BindingSetStringConverter();
- BindingSet bindingSet = converter.convert(bindingSetString, varOrder);
-
+ final VariableOrder varOrder = new VariableOrder("x", "a", "y", "b");
+ final BindingSetConverter<String> converter = new BindingSetStringConverter();
+ final BindingSet bindingSet = converter.convert(bindingSetString, varOrder);
+
// Ensure it converted to the expected reuslt.
final MapBindingSet expected = new MapBindingSet();
expected.addBinding("x", new URIImpl("http://value 1"));
expected.addBinding("y", new URIImpl("http://value 2"));
-
+
assertEquals(expected, bindingSet);
}
-
+
@Test
public void fromString_Decimal() throws BindingSetConversionException {
// Setup the String that will be converted.
final String bindingSetString = "2.5<<~>>http://www.w3.org/2001/XMLSchema#decimal";
// Convert it to a BindingSet
- BindingSetConverter<String> converter = new BindingSetStringConverter();
+ final BindingSetConverter<String> converter = new BindingSetStringConverter();
final BindingSet bindingSet = converter.convert(bindingSetString, new VariableOrder("x"));
// Ensure it converted to the expected result.
@@ -252,7 +252,7 @@ public class BindingSetStringConverterTest {
final String bindingSetString = "true<<~>>http://www.w3.org/2001/XMLSchema#boolean";
// Convert it to a BindingSet
- BindingSetConverter<String> converter = new BindingSetStringConverter();
+ final BindingSetConverter<String> converter = new BindingSetStringConverter();
final BindingSet bindingSet = converter.convert(bindingSetString, new VariableOrder("x"));
// Ensure it converted to the expected result.
@@ -268,7 +268,7 @@ public class BindingSetStringConverterTest {
final String bindingSetString = "5<<~>>http://www.w3.org/2001/XMLSchema#integer";
// Convert it to a BindingSet
- BindingSetConverter<String> converter = new BindingSetStringConverter();
+ final BindingSetConverter<String> converter = new BindingSetStringConverter();
final BindingSet bindingSet = converter.convert(bindingSetString, new VariableOrder("x"));
// Ensure it converted to the expected result.
@@ -286,10 +286,10 @@ public class BindingSetStringConverterTest {
"http://b<<~>>http://www.w3.org/2001/XMLSchema#anyURI";
// This variable order is too short.
- VariableOrder varOrder = new VariableOrder("x");
+ final VariableOrder varOrder = new VariableOrder("x");
// The conversion should throw an exception.
- BindingSetConverter<String> converter = new BindingSetStringConverter();
+ final BindingSetConverter<String> converter = new BindingSetStringConverter();
converter.convert(bindingSetString, varOrder);
}
@@ -301,10 +301,10 @@ public class BindingSetStringConverterTest {
"http://b<<~>>http://www.w3.org/2001/XMLSchema#anyURI";
// This variable order is too long.
- VariableOrder varOrder = new VariableOrder("x", "y", "z");
+ final VariableOrder varOrder = new VariableOrder("x", "y", "z");
// The conversion should throw an exception.
- BindingSetConverter<String> converter = new BindingSetStringConverter();
+ final BindingSetConverter<String> converter = new BindingSetStringConverter();
converter.convert(bindingSetString, varOrder);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/indexing/src/test/java/mvm/rya/indexing/external/tupleSet/PcjTablesIntegrationTests.java
----------------------------------------------------------------------
diff --git a/extras/indexing/src/test/java/mvm/rya/indexing/external/tupleSet/PcjTablesIntegrationTests.java b/extras/indexing/src/test/java/mvm/rya/indexing/external/tupleSet/PcjTablesIntegrationTests.java
index bb21d33..c452aad 100644
--- a/extras/indexing/src/test/java/mvm/rya/indexing/external/tupleSet/PcjTablesIntegrationTests.java
+++ b/extras/indexing/src/test/java/mvm/rya/indexing/external/tupleSet/PcjTablesIntegrationTests.java
@@ -8,9 +8,9 @@ package mvm.rya.indexing.external.tupleSet;
* 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
@@ -28,21 +28,6 @@ import java.util.HashSet;
import java.util.Map.Entry;
import java.util.Set;
-import mvm.rya.accumulo.AccumuloRdfConfiguration;
-import mvm.rya.accumulo.AccumuloRyaDAO;
-import mvm.rya.api.RdfCloudTripleStoreConfiguration;
-import mvm.rya.api.resolver.RyaTypeResolverException;
-import mvm.rya.indexing.accumulo.ConfigUtils;
-import mvm.rya.indexing.external.tupleSet.BindingSetConverter.BindingSetConversionException;
-import mvm.rya.indexing.external.tupleSet.PcjTables.PcjException;
-import mvm.rya.indexing.external.tupleSet.PcjTables.PcjMetadata;
-import mvm.rya.indexing.external.tupleSet.PcjTables.PcjTableNameFactory;
-import mvm.rya.indexing.external.tupleSet.PcjTables.PcjVarOrderFactory;
-import mvm.rya.indexing.external.tupleSet.PcjTables.ShiftVarOrderFactory;
-import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
-import mvm.rya.rdftriplestore.RdfCloudTripleStore;
-import mvm.rya.rdftriplestore.RyaSailRepository;
-
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.Connector;
@@ -77,6 +62,21 @@ import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import com.google.common.io.Files;
+import mvm.rya.accumulo.AccumuloRdfConfiguration;
+import mvm.rya.accumulo.AccumuloRyaDAO;
+import mvm.rya.api.RdfCloudTripleStoreConfiguration;
+import mvm.rya.indexing.accumulo.ConfigUtils;
+import mvm.rya.indexing.accumulo.VisibilityBindingSet;
+import mvm.rya.indexing.external.tupleSet.BindingSetConverter.BindingSetConversionException;
+import mvm.rya.indexing.external.tupleSet.PcjTables.PcjException;
+import mvm.rya.indexing.external.tupleSet.PcjTables.PcjMetadata;
+import mvm.rya.indexing.external.tupleSet.PcjTables.PcjTableNameFactory;
+import mvm.rya.indexing.external.tupleSet.PcjTables.PcjVarOrderFactory;
+import mvm.rya.indexing.external.tupleSet.PcjTables.ShiftVarOrderFactory;
+import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
+import mvm.rya.rdftriplestore.RdfCloudTripleStore;
+import mvm.rya.rdftriplestore.RyaSailRepository;
+
/**
* Performs integration test using {@link MiniAccumuloCluster} to ensure the
* functions of {@link PcjTables} work within a cluster setting.
@@ -85,7 +85,7 @@ public class PcjTablesIntegrationTests {
private static final Logger log = Logger.getLogger(PcjTablesIntegrationTests.class);
private static final AccumuloPcjSerializer converter = new AccumuloPcjSerializer();
-
+
protected static final String RYA_TABLE_PREFIX = "demo_";
// Rya data store and connections.
@@ -122,8 +122,8 @@ public class PcjTablesIntegrationTests {
// Create a PCJ table in the Mini Accumulo.
final String pcjTableName = new PcjTableNameFactory().makeTableName(RYA_TABLE_PREFIX, "testPcj");
- Set<VariableOrder> varOrders = new ShiftVarOrderFactory().makeVarOrders(new VariableOrder("name;age"));
- PcjTables pcjs = new PcjTables();
+ final Set<VariableOrder> varOrders = new ShiftVarOrderFactory().makeVarOrders(new VariableOrder("name;age"));
+ final PcjTables pcjs = new PcjTables();
pcjs.createPcjTable(accumuloConn, pcjTableName, varOrders, sparql);
// Fetch the PcjMetadata and ensure it has the correct values.
@@ -151,35 +151,38 @@ public class PcjTablesIntegrationTests {
// Create a PCJ table in the Mini Accumulo.
final String pcjTableName = new PcjTableNameFactory().makeTableName(RYA_TABLE_PREFIX, "testPcj");
- Set<VariableOrder> varOrders = new ShiftVarOrderFactory().makeVarOrders(new VariableOrder("name;age"));
- PcjTables pcjs = new PcjTables();
+ final Set<VariableOrder> varOrders = new ShiftVarOrderFactory().makeVarOrders(new VariableOrder("name;age"));
+ final PcjTables pcjs = new PcjTables();
pcjs.createPcjTable(accumuloConn, pcjTableName, varOrders, sparql);
// Add a few results to the PCJ table.
- MapBindingSet alice = new MapBindingSet();
+ final MapBindingSet alice = new MapBindingSet();
alice.addBinding("name", new URIImpl("http://Alice"));
alice.addBinding("age", new NumericLiteralImpl(14, XMLSchema.INTEGER));
- MapBindingSet bob = new MapBindingSet();
+ final MapBindingSet bob = new MapBindingSet();
bob.addBinding("name", new URIImpl("http://Bob"));
bob.addBinding("age", new NumericLiteralImpl(16, XMLSchema.INTEGER));
- MapBindingSet charlie = new MapBindingSet();
+ final MapBindingSet charlie = new MapBindingSet();
charlie.addBinding("name", new URIImpl("http://Charlie"));
charlie.addBinding("age", new NumericLiteralImpl(12, XMLSchema.INTEGER));
- Set<BindingSet> results = Sets.<BindingSet>newHashSet(alice, bob, charlie);
- pcjs.addResults(accumuloConn, pcjTableName, results);
+ final Set<BindingSet> results = Sets.<BindingSet>newHashSet(alice, bob, charlie);
+ pcjs.addResults(accumuloConn, pcjTableName, Sets.<VisibilityBindingSet>newHashSet(
+ new VisibilityBindingSet(alice),
+ new VisibilityBindingSet(bob),
+ new VisibilityBindingSet(charlie)));
// Make sure the cardinality was updated.
- PcjMetadata metadata = pcjs.getPcjMetadata(accumuloConn, pcjTableName);
+ final PcjMetadata metadata = pcjs.getPcjMetadata(accumuloConn, pcjTableName);
assertEquals(3, metadata.getCardinality());
// Scan Accumulo for the stored results.
- Multimap<String, BindingSet> fetchedResults = loadPcjResults(accumuloConn, pcjTableName);
+ final Multimap<String, BindingSet> fetchedResults = loadPcjResults(accumuloConn, pcjTableName);
// Ensure the expected results match those that were stored.
- Multimap<String, BindingSet> expectedResults = HashMultimap.create();
+ final Multimap<String, BindingSet> expectedResults = HashMultimap.create();
expectedResults.putAll("name;age", results);
expectedResults.putAll("age;name", results);
assertEquals(expectedResults, fetchedResults);
@@ -194,7 +197,7 @@ public class PcjTablesIntegrationTests {
@Test
public void populatePcj() throws RepositoryException, PcjException, TableNotFoundException, BindingSetConversionException {
// Load some Triples into Rya.
- Set<Statement> triples = new HashSet<>();
+ final Set<Statement> triples = new HashSet<>();
triples.add( new StatementImpl(new URIImpl("http://Alice"), new URIImpl("http://hasAge"), new NumericLiteralImpl(14, XMLSchema.INTEGER)) );
triples.add( new StatementImpl(new URIImpl("http://Alice"), new URIImpl("http://playsSport"), new LiteralImpl("Soccer")) );
triples.add( new StatementImpl(new URIImpl("http://Bob"), new URIImpl("http://hasAge"), new NumericLiteralImpl(16, XMLSchema.INTEGER)) );
@@ -204,7 +207,7 @@ public class PcjTablesIntegrationTests {
triples.add( new StatementImpl(new URIImpl("http://Eve"), new URIImpl("http://hasAge"), new NumericLiteralImpl(43, XMLSchema.INTEGER)) );
triples.add( new StatementImpl(new URIImpl("http://Eve"), new URIImpl("http://playsSport"), new LiteralImpl("Soccer")) );
- for(Statement triple : triples) {
+ for(final Statement triple : triples) {
ryaConn.add(triple);
}
@@ -218,36 +221,36 @@ public class PcjTablesIntegrationTests {
"}";
final String pcjTableName = new PcjTableNameFactory().makeTableName(RYA_TABLE_PREFIX, "testPcj");
- Set<VariableOrder> varOrders = new ShiftVarOrderFactory().makeVarOrders(new VariableOrder("name;age"));
- PcjTables pcjs = new PcjTables();
+ final Set<VariableOrder> varOrders = new ShiftVarOrderFactory().makeVarOrders(new VariableOrder("name;age"));
+ final PcjTables pcjs = new PcjTables();
pcjs.createPcjTable(accumuloConn, pcjTableName, varOrders, sparql);
// Populate the PCJ table using a Rya connection.
pcjs.populatePcj(accumuloConn, pcjTableName, ryaConn);
// Scan Accumulo for the stored results.
- Multimap<String, BindingSet> fetchedResults = loadPcjResults(accumuloConn, pcjTableName);
+ final Multimap<String, BindingSet> fetchedResults = loadPcjResults(accumuloConn, pcjTableName);
// Make sure the cardinality was updated.
- PcjMetadata metadata = pcjs.getPcjMetadata(accumuloConn, pcjTableName);
+ final PcjMetadata metadata = pcjs.getPcjMetadata(accumuloConn, pcjTableName);
assertEquals(3, metadata.getCardinality());
// Ensure the expected results match those that were stored.
- MapBindingSet alice = new MapBindingSet();
+ final MapBindingSet alice = new MapBindingSet();
alice.addBinding("name", new URIImpl("http://Alice"));
alice.addBinding("age", new NumericLiteralImpl(14, XMLSchema.INTEGER));
- MapBindingSet bob = new MapBindingSet();
+ final MapBindingSet bob = new MapBindingSet();
bob.addBinding("name", new URIImpl("http://Bob"));
bob.addBinding("age", new NumericLiteralImpl(16, XMLSchema.INTEGER));
- MapBindingSet charlie = new MapBindingSet();
+ final MapBindingSet charlie = new MapBindingSet();
charlie.addBinding("name", new URIImpl("http://Charlie"));
charlie.addBinding("age", new NumericLiteralImpl(12, XMLSchema.INTEGER));
- Set<BindingSet> results = Sets.<BindingSet>newHashSet(alice, bob, charlie);
+ final Set<BindingSet> results = Sets.<BindingSet>newHashSet(alice, bob, charlie);
- Multimap<String, BindingSet> expectedResults = HashMultimap.create();
+ final Multimap<String, BindingSet> expectedResults = HashMultimap.create();
expectedResults.putAll("name;age", results);
expectedResults.putAll("age;name", results);
assertEquals(expectedResults, fetchedResults);
@@ -262,7 +265,7 @@ public class PcjTablesIntegrationTests {
@Test
public void createAndPopulatePcj() throws RepositoryException, PcjException, TableNotFoundException, BindingSetConversionException {
// Load some Triples into Rya.
- Set<Statement> triples = new HashSet<>();
+ final Set<Statement> triples = new HashSet<>();
triples.add( new StatementImpl(new URIImpl("http://Alice"), new URIImpl("http://hasAge"), new NumericLiteralImpl(14, XMLSchema.INTEGER)) );
triples.add( new StatementImpl(new URIImpl("http://Alice"), new URIImpl("http://playsSport"), new LiteralImpl("Soccer")) );
triples.add( new StatementImpl(new URIImpl("http://Bob"), new URIImpl("http://hasAge"), new NumericLiteralImpl(16, XMLSchema.INTEGER)) );
@@ -272,7 +275,7 @@ public class PcjTablesIntegrationTests {
triples.add( new StatementImpl(new URIImpl("http://Eve"), new URIImpl("http://hasAge"), new NumericLiteralImpl(43, XMLSchema.INTEGER)) );
triples.add( new StatementImpl(new URIImpl("http://Eve"), new URIImpl("http://playsSport"), new LiteralImpl("Soccer")) );
- for(Statement triple : triples) {
+ for(final Statement triple : triples) {
ryaConn.add(triple);
}
@@ -288,32 +291,32 @@ public class PcjTablesIntegrationTests {
final String pcjTableName = new PcjTableNameFactory().makeTableName(RYA_TABLE_PREFIX, "testPcj");
// Create and populate the PCJ table.
- PcjTables pcjs = new PcjTables();
+ final PcjTables pcjs = new PcjTables();
pcjs.createAndPopulatePcj(ryaConn, accumuloConn, pcjTableName, sparql, new String[]{"name", "age"}, Optional.<PcjVarOrderFactory>absent());
// Make sure the cardinality was updated.
- PcjMetadata metadata = pcjs.getPcjMetadata(accumuloConn, pcjTableName);
+ final PcjMetadata metadata = pcjs.getPcjMetadata(accumuloConn, pcjTableName);
assertEquals(3, metadata.getCardinality());
// Scan Accumulo for the stored results.
- Multimap<String, BindingSet> fetchedResults = loadPcjResults(accumuloConn, pcjTableName);
+ final Multimap<String, BindingSet> fetchedResults = loadPcjResults(accumuloConn, pcjTableName);
// Ensure the expected results match those that were stored.
- MapBindingSet alice = new MapBindingSet();
+ final MapBindingSet alice = new MapBindingSet();
alice.addBinding("name", new URIImpl("http://Alice"));
alice.addBinding("age", new NumericLiteralImpl(14, XMLSchema.INTEGER));
- MapBindingSet bob = new MapBindingSet();
+ final MapBindingSet bob = new MapBindingSet();
bob.addBinding("name", new URIImpl("http://Bob"));
bob.addBinding("age", new NumericLiteralImpl(16, XMLSchema.INTEGER));
- MapBindingSet charlie = new MapBindingSet();
+ final MapBindingSet charlie = new MapBindingSet();
charlie.addBinding("name", new URIImpl("http://Charlie"));
charlie.addBinding("age", new NumericLiteralImpl(12, XMLSchema.INTEGER));
- Set<BindingSet> results = Sets.<BindingSet>newHashSet(alice, bob, charlie);
+ final Set<BindingSet> results = Sets.<BindingSet>newHashSet(alice, bob, charlie);
- Multimap<String, BindingSet> expectedResults = HashMultimap.create();
+ final Multimap<String, BindingSet> expectedResults = HashMultimap.create();
expectedResults.putAll("name;age", results);
expectedResults.putAll("age;name", results);
@@ -325,21 +328,21 @@ public class PcjTablesIntegrationTests {
* multimap stores a set of deserialized binding sets that were in the PCJ
* table for every variable order that is found in the PCJ metadata.
*/
- private static Multimap<String, BindingSet> loadPcjResults(Connector accumuloConn, String pcjTableName) throws PcjException, TableNotFoundException, BindingSetConversionException {
- Multimap<String, BindingSet> fetchedResults = HashMultimap.create();
+ private static Multimap<String, BindingSet> loadPcjResults(final Connector accumuloConn, final String pcjTableName) throws PcjException, TableNotFoundException, BindingSetConversionException {
+ final Multimap<String, BindingSet> fetchedResults = HashMultimap.create();
// Get the variable orders the data was written to.
- PcjTables pcjs = new PcjTables();
- PcjMetadata pcjMetadata = pcjs.getPcjMetadata(accumuloConn, pcjTableName);
+ final PcjTables pcjs = new PcjTables();
+ final PcjMetadata pcjMetadata = pcjs.getPcjMetadata(accumuloConn, pcjTableName);
// Scan Accumulo for the stored results.
- for(VariableOrder varOrder : pcjMetadata.getVarOrders()) {
- Scanner scanner = accumuloConn.createScanner(pcjTableName, new Authorizations());
+ for(final VariableOrder varOrder : pcjMetadata.getVarOrders()) {
+ final Scanner scanner = accumuloConn.createScanner(pcjTableName, new Authorizations());
scanner.fetchColumnFamily( new Text(varOrder.toString()) );
- for(Entry<Key, Value> entry : scanner) {
- byte[] serializedResult = entry.getKey().getRow().getBytes();
- BindingSet result = converter.convert(serializedResult, varOrder);
+ for(final Entry<Key, Value> entry : scanner) {
+ final byte[] serializedResult = entry.getKey().getRow().getBytes();
+ final BindingSet result = converter.convert(serializedResult, varOrder);
fetchedResults.put(varOrder.toString(), result);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/indexing/src/test/java/mvm/rya/indexing/external/tupleSet/VisibilityBindingSetStringConverterTest.java
----------------------------------------------------------------------
diff --git a/extras/indexing/src/test/java/mvm/rya/indexing/external/tupleSet/VisibilityBindingSetStringConverterTest.java b/extras/indexing/src/test/java/mvm/rya/indexing/external/tupleSet/VisibilityBindingSetStringConverterTest.java
new file mode 100644
index 0000000..ccd27ac
--- /dev/null
+++ b/extras/indexing/src/test/java/mvm/rya/indexing/external/tupleSet/VisibilityBindingSetStringConverterTest.java
@@ -0,0 +1,132 @@
+/*
+ * 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 mvm.rya.indexing.external.tupleSet;
+
+import static mvm.rya.indexing.external.tupleSet.VisibilityBindingSetStringConverter.VISIBILITY_DELIM;
+import static org.junit.Assert.assertEquals;
+
+import org.junit.Test;
+import org.openrdf.model.impl.URIImpl;
+import org.openrdf.query.BindingSet;
+import org.openrdf.query.impl.MapBindingSet;
+
+import mvm.rya.indexing.accumulo.VisibilityBindingSet;
+import mvm.rya.indexing.external.tupleSet.BindingSetConverter.BindingSetConversionException;
+import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
+
+/**
+ * Tests the methods of {@link BindingSetStringConverter}.
+ */
+public class VisibilityBindingSetStringConverterTest {
+ @Test
+ public void toString_URIs() throws BindingSetConversionException {
+ // Setup the binding set that will be converted.
+ 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://c"));
+
+ final VisibilityBindingSet visiSet = new VisibilityBindingSet(originalBindingSet, "A&B&C");
+
+ // Convert it to a String.
+ final VariableOrder varOrder = new VariableOrder("y", "z", "x");
+ final VisibilityBindingSetStringConverter converter = new VisibilityBindingSetStringConverter();
+ final String bindingSetString = converter.convert(visiSet, varOrder);
+
+ // Ensure it converted to the expected result.l
+ final String expected =
+ "http://b<<~>>http://www.w3.org/2001/XMLSchema#anyURI:::" +
+ "http://c<<~>>http://www.w3.org/2001/XMLSchema#anyURI:::" +
+ "http://a<<~>>http://www.w3.org/2001/XMLSchema#anyURI" +
+ VISIBILITY_DELIM + "A&B&C";
+
+ assertEquals(expected, bindingSetString);
+ }
+
+ @Test
+ public void fromString() throws BindingSetConversionException {
+ // Setup the String that will be converted.
+ final String bindingSetString =
+ "http://b<<~>>http://www.w3.org/2001/XMLSchema#anyURI:::" +
+ "http://c<<~>>http://www.w3.org/2001/XMLSchema#anyURI:::" +
+ "http://a<<~>>http://www.w3.org/2001/XMLSchema#anyURI" +
+ VISIBILITY_DELIM + "A&B";
+
+ // Convert it to a BindingSet
+ final VariableOrder varOrder = new VariableOrder("y", "z", "x");
+ final VisibilityBindingSetStringConverter converter = new VisibilityBindingSetStringConverter();
+ final BindingSet bindingSet = converter.convert(bindingSetString, varOrder);
+
+ // Ensure it converted to the expected result.
+ final MapBindingSet expected = new MapBindingSet();
+ expected.addBinding("z", new URIImpl("http://c"));
+ expected.addBinding("y", new URIImpl("http://b"));
+ expected.addBinding("x", new URIImpl("http://a"));
+ final VisibilityBindingSet visiSet = new VisibilityBindingSet(expected, "A&B");
+
+ assertEquals(visiSet, bindingSet);
+ }
+
+ @Test
+ public void toString_URIs_noVisi() throws BindingSetConversionException {
+ // Setup the binding set that will be converted.
+ 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://c"));
+
+ final VisibilityBindingSet visiSet = new VisibilityBindingSet(originalBindingSet);
+
+ // Convert it to a String.
+ final VariableOrder varOrder = new VariableOrder("y", "z", "x");
+ final VisibilityBindingSetStringConverter converter = new VisibilityBindingSetStringConverter();
+ final String bindingSetString = converter.convert(visiSet, varOrder);
+
+ // Ensure it converted to the expected result.l
+ final String expected =
+ "http://b<<~>>http://www.w3.org/2001/XMLSchema#anyURI:::" +
+ "http://c<<~>>http://www.w3.org/2001/XMLSchema#anyURI:::" +
+ "http://a<<~>>http://www.w3.org/2001/XMLSchema#anyURI";
+
+ assertEquals(expected, bindingSetString);
+ }
+
+ @Test
+ public void fromString_noVisi() throws BindingSetConversionException {
+ // Setup the String that will be converted.
+ final String bindingSetString =
+ "http://b<<~>>http://www.w3.org/2001/XMLSchema#anyURI:::" +
+ "http://c<<~>>http://www.w3.org/2001/XMLSchema#anyURI:::" +
+ "http://a<<~>>http://www.w3.org/2001/XMLSchema#anyURI";
+
+ // Convert it to a BindingSet
+ final VariableOrder varOrder = new VariableOrder("y", "z", "x");
+ final VisibilityBindingSetStringConverter converter = new VisibilityBindingSetStringConverter();
+ final BindingSet bindingSet = converter.convert(bindingSetString, varOrder);
+
+ // Ensure it converted to the expected result.
+ final MapBindingSet expected = new MapBindingSet();
+ expected.addBinding("z", new URIImpl("http://c"));
+ expected.addBinding("y", new URIImpl("http://b"));
+ expected.addBinding("x", new URIImpl("http://a"));
+ final VisibilityBindingSet visiSet = new VisibilityBindingSet(expected);
+
+ assertEquals(visiSet, bindingSet);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/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 12e32b6..58e2f9a 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
@@ -232,7 +232,7 @@ public class CreatePcj {
checkNotNull(spMetadata);
checkNotNull(batch);
- BindingSetStringConverter converter = new BindingSetStringConverter();
+ final BindingSetStringConverter converter = new BindingSetStringConverter();
try(TypedTransaction tx = STRING_TYPED_LAYER.wrap(fluo.newTransaction())) {
// Get the node's variable order.
@@ -240,13 +240,13 @@ public class CreatePcj {
final VariableOrder varOrder = spMetadata.getVariableOrder();
for(final BindingSet bindingSet : batch) {
- MapBindingSet spBindingSet = new MapBindingSet();
- for(String var : varOrder) {
- Binding binding = bindingSet.getBinding(var);
+ final MapBindingSet spBindingSet = new MapBindingSet();
+ for(final String var : varOrder) {
+ final Binding binding = bindingSet.getBinding(var);
spBindingSet.addBinding(binding);
}
- String bindingSetStr = converter.convert(spBindingSet, varOrder);
+ final String bindingSetStr = converter.convert(spBindingSet, varOrder);
// Write the binding set entry to Fluo for the statement pattern.
tx.mutate().row(spNodeId + NODEID_BS_DELIM + bindingSetStr)
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.api/src/main/java/org/apache/rya/indexing/pcj/fluo/api/InsertTriples.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.api/src/main/java/org/apache/rya/indexing/pcj/fluo/api/InsertTriples.java b/extras/rya.pcj.fluo/pcj.fluo.api/src/main/java/org/apache/rya/indexing/pcj/fluo/api/InsertTriples.java
index 02e871f..092ad9c 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.api/src/main/java/org/apache/rya/indexing/pcj/fluo/api/InsertTriples.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.api/src/main/java/org/apache/rya/indexing/pcj/fluo/api/InsertTriples.java
@@ -28,7 +28,11 @@ import org.apache.log4j.Logger;
import org.apache.rya.indexing.pcj.fluo.app.StringTypeLayer;
import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryColumns;
+import com.google.common.base.Optional;
+
import io.fluo.api.client.FluoClient;
+import io.fluo.api.types.Encoder;
+import io.fluo.api.types.StringEncoder;
import io.fluo.api.types.TypedTransaction;
import mvm.rya.api.RdfCloudTripleStoreConstants.TABLE_LAYOUT;
import mvm.rya.api.domain.RyaStatement;
@@ -53,14 +57,17 @@ public class InsertTriples {
*/
private static final WholeRowTripleResolver TRIPLE_RESOLVER = new WholeRowTripleResolver();
+ private static final Encoder ENCODER = new StringEncoder();
+
/**
* Inserts a triple into Fluo.
*
* @param fluo - A connection to the Fluo table that will be updated. (not null)
* @param triple - The triple to insert. (not null)
+ * @param visibility - The visibility/permissions required to view this triple once stored. (not null)
*/
- public void insert(final FluoClient fluo, final RyaStatement triple) {
- insert(fluo, Collections.singleton(triple));
+ public void insert(final FluoClient fluo, final RyaStatement triple, final Optional<String> visibility) {
+ insert(fluo, Collections.singleton(triple), visibility);
}
/**
@@ -68,15 +75,18 @@ public class InsertTriples {
*
* @param fluo - A connection to the Fluo table that will be updated. (not null)
* @param triples - The triples to insert. (not null)
+ * @param visibility - The visibility/permissions required to view the triples once stored.
+ * Note: The same visibility will be applied to each triple.(not null)
*/
- public void insert(final FluoClient fluo, final Collection<RyaStatement> triples) {
+ public void insert(final FluoClient fluo, final Collection<RyaStatement> triples, final Optional<String> visibility) {
checkNotNull(fluo);
checkNotNull(triples);
+ checkNotNull(visibility);
try(TypedTransaction tx = STRING_TYPED_LAYER.wrap(fluo.newTransaction())) {
for(final RyaStatement triple : triples) {
try {
- tx.mutate().row(spoFormat(triple)).col(FluoQueryColumns.TRIPLES).set();
+ tx.mutate().row(spoFormat(triple)).col(FluoQueryColumns.TRIPLES).set(ENCODER.encode(visibility.or("")));
} catch (final TripleRowResolverException e) {
log.error("Could not convert a Triple into the SPO format: " + triple);
}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/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 3af079f..fc88a3c 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
@@ -50,8 +50,10 @@ import io.fluo.api.data.Bytes;
import io.fluo.api.data.Column;
import io.fluo.api.types.Encoder;
import io.fluo.api.types.StringEncoder;
+import mvm.rya.indexing.accumulo.VisibilityBindingSet;
import mvm.rya.indexing.external.tupleSet.BindingSetStringConverter;
import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
+import mvm.rya.indexing.external.tupleSet.VisibilityBindingSetStringConverter;
/**
* Updates the results of a Filter node when its child has added a new Binding
@@ -60,9 +62,10 @@ import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
@ParametersAreNonnullByDefault
public class FilterResultUpdater {
- private final Encoder encoder = new StringEncoder();
+ private static final Encoder ENCODER = new StringEncoder();
- private final BindingSetStringConverter converter = new BindingSetStringConverter();
+ private static final BindingSetStringConverter ID_CONVERTER = new BindingSetStringConverter();
+ private static final VisibilityBindingSetStringConverter VALUE_CONVERTER = new VisibilityBindingSetStringConverter();
/**
* A utility class used to search SPARQL queries for Filters.
@@ -102,7 +105,7 @@ public class FilterResultUpdater {
*/
public void updateFilterResults(
final TransactionBase tx,
- final BindingSet childBindingSet,
+ final VisibilityBindingSet childBindingSet,
final FilterMetadata filterMetadata) throws Exception {
checkNotNull(tx);
checkNotNull(childBindingSet);
@@ -126,11 +129,14 @@ public class FilterResultUpdater {
filterBindingSet.addBinding(binding);
}
}
- final String filterBindingSetString = converter.convert(filterBindingSet, filterVarOrder);
- final Bytes row = encoder.encode( filterMetadata.getNodeId() + NODEID_BS_DELIM + filterBindingSetString );
+ final String filterBindingSetIdString = ID_CONVERTER.convert(filterBindingSet, filterVarOrder);
+ String filterBindingSetValueString = "";
+ filterBindingSetValueString = VALUE_CONVERTER.convert(childBindingSet, filterVarOrder);
+
+ final Bytes row = ENCODER.encode( filterMetadata.getNodeId() + NODEID_BS_DELIM + filterBindingSetIdString );
final Column col = FluoQueryColumns.FILTER_BINDING_SET;
- final Bytes value = encoder.encode(filterBindingSetString);
+ final Bytes value = ENCODER.encode(filterBindingSetValueString);
tx.set(row, col, value);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/IncUpdateDAO.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/IncUpdateDAO.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/IncUpdateDAO.java
index 5cb9869..5bc9d0a 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/IncUpdateDAO.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/IncUpdateDAO.java
@@ -44,7 +44,7 @@ public class IncUpdateDAO {
private static final StringTypeLayer stl = new StringTypeLayer();
private static final WholeRowTripleResolver tr = new WholeRowTripleResolver();
- private static RyaStatement deserializeTriple(final Bytes row) {
+ public static RyaStatement deserializeTriple(final Bytes row) {
final byte[] rowArray = row.toArray();
RyaStatement rs = null;
[2/3] incubator-rya git commit: RYA-53 Added visibility support
Posted by pu...@apache.org.
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/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 bc558a5..700d0fb 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
@@ -37,6 +37,7 @@ import org.openrdf.query.Binding;
import org.openrdf.query.BindingSet;
import org.openrdf.query.impl.MapBindingSet;
+import com.google.common.base.Joiner;
import com.google.common.base.Optional;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
@@ -50,10 +51,11 @@ import io.fluo.api.iterator.ColumnIterator;
import io.fluo.api.iterator.RowIterator;
import io.fluo.api.types.Encoder;
import io.fluo.api.types.StringEncoder;
-import mvm.rya.indexing.external.tupleSet.BindingSetConverter;
+import mvm.rya.indexing.accumulo.VisibilityBindingSet;
import mvm.rya.indexing.external.tupleSet.BindingSetConverter.BindingSetConversionException;
import mvm.rya.indexing.external.tupleSet.BindingSetStringConverter;
import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
+import mvm.rya.indexing.external.tupleSet.VisibilityBindingSetStringConverter;
/**
* Updates the results of a Join node when one of its children has added a
@@ -62,7 +64,8 @@ import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
@ParametersAreNonnullByDefault
public class JoinResultUpdater {
- private static final BindingSetConverter<String> converter = new BindingSetStringConverter();
+ private static final BindingSetStringConverter idConverter = new BindingSetStringConverter();
+ private static final VisibilityBindingSetStringConverter valueConverter = new VisibilityBindingSetStringConverter();
private final FluoQueryMetadataDAO queryDao = new FluoQueryMetadataDAO();
private final Encoder encoder = new StringEncoder();
@@ -80,7 +83,7 @@ public class JoinResultUpdater {
public void updateJoinResults(
final TransactionBase tx,
final String childId,
- final BindingSet childBindingSet,
+ final VisibilityBindingSet childBindingSet,
final JoinMetadata joinMetadata) throws BindingSetConversionException {
checkNotNull(tx);
checkNotNull(childId);
@@ -113,10 +116,10 @@ public class JoinResultUpdater {
}
// Iterates over the sibling node's BindingSets that join with the new binding set.
- FluoTableIterator siblingBindingSets = makeSiblingScanIterator(childId, childBindingSet, siblingId, tx);
+ final FluoTableIterator siblingBindingSets = makeSiblingScanIterator(childId, childBindingSet, siblingId, tx);
// Iterates over the resulting BindingSets from the join.
- final Iterator<BindingSet> newJoinResults;
+ final Iterator<VisibilityBindingSet> newJoinResults;
if(emittingSide == Side.LEFT) {
newJoinResults = joinAlgorithm.newLeftResult(childBindingSet, siblingBindingSets);
} else {
@@ -124,14 +127,15 @@ public class JoinResultUpdater {
}
// Insert the new join binding sets to the Fluo table.
- VariableOrder joinVarOrder = joinMetadata.getVariableOrder();
+ final VariableOrder joinVarOrder = joinMetadata.getVariableOrder();
while(newJoinResults.hasNext()) {
- BindingSet newJoinResult = newJoinResults.next();
- String joinBindingSetString = converter.convert(newJoinResult, joinVarOrder);
+ final BindingSet newJoinResult = newJoinResults.next();
+ final String joinBindingSetStringId = idConverter.convert(newJoinResult, joinVarOrder);
+ final String joinBindingSetStringValue = valueConverter.convert(newJoinResult, joinVarOrder);
- final Bytes row = encoder.encode(joinMetadata.getNodeId() + NODEID_BS_DELIM + joinBindingSetString);
+ final Bytes row = encoder.encode(joinMetadata.getNodeId() + NODEID_BS_DELIM + joinBindingSetStringId);
final Column col = FluoQueryColumns.JOIN_BINDING_SET;
- final Bytes value = encoder.encode(joinBindingSetString);
+ final Bytes value = encoder.encode(joinBindingSetStringValue);
tx.set(row, col, value);
}
}
@@ -143,14 +147,14 @@ public class JoinResultUpdater {
LEFT, RIGHT;
}
- private FluoTableIterator makeSiblingScanIterator(String childId, BindingSet childBindingSet, String siblingId, TransactionBase tx) throws BindingSetConversionException {
+ private FluoTableIterator makeSiblingScanIterator(final String childId, final BindingSet childBindingSet, final String siblingId, final TransactionBase tx) throws BindingSetConversionException {
// Get the common variable orders. These are used to build the prefix.
final VariableOrder childVarOrder = getVarOrder(tx, childId);
final VariableOrder siblingVarOrder = getVarOrder(tx, siblingId);
- List<String> commonVars = getCommonVars(childVarOrder, siblingVarOrder);
+ final List<String> commonVars = getCommonVars(childVarOrder, siblingVarOrder);
// Get the Binding strings
- final String childBindingSetString = converter.convert(childBindingSet, childVarOrder);
+ final String childBindingSetString = valueConverter.convert(childBindingSet, childVarOrder);
final String[] childBindingStrings = FluoStringConverter.toBindingStrings(childBindingSetString);
// Create the prefix that will be used to scan for binding sets of the sibling node.
@@ -285,20 +289,20 @@ public class JoinResultUpdater {
public static interface IterativeJoin {
/**
- * Invoked when a new {@link BindingSet} is emitted from the left child
+ * Invoked when a new {@link VisibilityBindingSet} is emitted from the left child
* node of the join. The Fluo table is scanned for results on the right
* side that will be joined with the new result.
*
- * @param newLeftResult - A new BindingSet that has been emitted from
+ * @param newLeftResult - A new VisibilityBindingSet that has been emitted from
* the left child node.
* @param rightResults - The right child node's binding sets that will
* be joined with the new left result. (not null)
* @return The new BindingSet results for the join.
*/
- public Iterator<BindingSet> newLeftResult(BindingSet newLeftResult, Iterator<BindingSet> rightResults);
+ public Iterator<VisibilityBindingSet> newLeftResult(VisibilityBindingSet newLeftResult, Iterator<VisibilityBindingSet> rightResults);
/**
- * Invoked when a new {@link BindingSet} is emitted from the right child
+ * Invoked when a new {@link VisibilityBindingSet} is emitted from the right child
* node of the join. The Fluo table is scanned for results on the left
* side that will be joined with the new result.
*
@@ -308,7 +312,7 @@ public class JoinResultUpdater {
* the right child node.
* @return The new BindingSet results for the join.
*/
- public Iterator<BindingSet> newRightResult(Iterator<BindingSet> leftResults, BindingSet newRightResult);
+ public Iterator<VisibilityBindingSet> newRightResult(Iterator<VisibilityBindingSet> leftResults, VisibilityBindingSet newRightResult);
}
/**
@@ -321,7 +325,7 @@ public class JoinResultUpdater {
*/
public static final class NaturalJoin implements IterativeJoin {
@Override
- public Iterator<BindingSet> newLeftResult(BindingSet newLeftResult, Iterator<BindingSet> rightResults) {
+ public Iterator<VisibilityBindingSet> newLeftResult(final VisibilityBindingSet newLeftResult, final Iterator<VisibilityBindingSet> rightResults) {
checkNotNull(newLeftResult);
checkNotNull(rightResults);
@@ -330,7 +334,7 @@ public class JoinResultUpdater {
}
@Override
- public Iterator<BindingSet> newRightResult(Iterator<BindingSet> leftResults, BindingSet newRightResult) {
+ public Iterator<VisibilityBindingSet> newRightResult(final Iterator<VisibilityBindingSet> leftResults, final VisibilityBindingSet newRightResult) {
checkNotNull(leftResults);
checkNotNull(newRightResult);
@@ -349,14 +353,14 @@ public class JoinResultUpdater {
*/
public static final class LeftOuterJoin implements IterativeJoin {
@Override
- public Iterator<BindingSet> newLeftResult(BindingSet newLeftResult, Iterator<BindingSet> rightResults) {
+ public Iterator<VisibilityBindingSet> newLeftResult(final VisibilityBindingSet newLeftResult, final Iterator<VisibilityBindingSet> rightResults) {
checkNotNull(newLeftResult);
checkNotNull(rightResults);
// If the required portion does not join with any optional portions,
// then emit a BindingSet that matches the new left result.
if(!rightResults.hasNext()) {
- return Lists.<BindingSet>newArrayList(newLeftResult).iterator();
+ return Lists.<VisibilityBindingSet>newArrayList(newLeftResult).iterator();
}
// Otherwise, return an iterator that holds the new required result
@@ -365,7 +369,7 @@ public class JoinResultUpdater {
}
@Override
- public Iterator<BindingSet> newRightResult(final Iterator<BindingSet> leftResults, final BindingSet newRightResult) {
+ public Iterator<VisibilityBindingSet> newRightResult(final Iterator<VisibilityBindingSet> leftResults, final VisibilityBindingSet newRightResult) {
checkNotNull(leftResults);
checkNotNull(newRightResult);
@@ -382,10 +386,10 @@ public class JoinResultUpdater {
* This is done lazily so that you don't have to load all of the BindingSets
* into memory at once.
*/
- private static final class LazyJoiningIterator implements Iterator<BindingSet> {
+ private static final class LazyJoiningIterator implements Iterator<VisibilityBindingSet> {
- private final BindingSet newResult;
- private final Iterator<BindingSet> joinedResults;
+ private final VisibilityBindingSet newResult;
+ private final Iterator<VisibilityBindingSet> joinedResults;
/**
* Constructs an instance of {@link LazyJoiningIterator}.
@@ -393,9 +397,9 @@ public class JoinResultUpdater {
* @param newResult - A binding set that will be joined with some other binding sets. (not null)
* @param joinResults - The binding sets that will be joined with {@code newResult}. (not null)
*/
- public LazyJoiningIterator(BindingSet newResult, Iterator<BindingSet> joinResults) {
+ public LazyJoiningIterator(final VisibilityBindingSet newResult, final Iterator<VisibilityBindingSet> joinResults) {
this.newResult = checkNotNull(newResult);
- this.joinedResults = checkNotNull(joinResults);
+ joinedResults = checkNotNull(joinResults);
}
@Override
@@ -404,18 +408,28 @@ public class JoinResultUpdater {
}
@Override
- public BindingSet next() {
+ public VisibilityBindingSet next() {
final MapBindingSet bs = new MapBindingSet();
- for(Binding binding : newResult) {
+ for(final Binding binding : newResult) {
bs.addBinding(binding);
}
- for(Binding binding : joinedResults.next()) {
+ final VisibilityBindingSet joinResult = joinedResults.next();
+ for(final Binding binding : joinResult) {
bs.addBinding(binding);
}
- return bs;
+ String visibility = "";
+ final Joiner join = Joiner.on(")&(");
+ final String leftVisi = newResult.getVisibility();
+ final String rightVisi = joinResult.getVisibility();
+ if(leftVisi.isEmpty() || rightVisi.isEmpty()) {
+ visibility = (leftVisi + rightVisi).trim();
+ } else {
+ visibility = "(" + join.join(leftVisi, rightVisi) + ")";
+ }
+ return new VisibilityBindingSet(bs, visibility);
}
@Override
@@ -428,7 +442,7 @@ public class JoinResultUpdater {
* Iterates over rows that have a Binding Set column and returns the unmarshalled
* {@link BindingSet}s.
*/
- private static final class FluoTableIterator implements Iterator<BindingSet> {
+ private static final class FluoTableIterator implements Iterator<VisibilityBindingSet> {
private static final Set<Column> BINDING_SET_COLUMNS = Sets.newHashSet(
FluoQueryColumns.STATEMENT_PATTERN_BINDING_SET,
@@ -445,7 +459,7 @@ public class JoinResultUpdater {
* @param varOrder - The Variable Order of binding sets that will be
* read from the Fluo Table. (not null)
*/
- public FluoTableIterator(RowIterator rows, VariableOrder varOrder) {
+ public FluoTableIterator(final RowIterator rows, final VariableOrder varOrder) {
this.rows = checkNotNull(rows);
this.varOrder = checkNotNull(varOrder);
}
@@ -456,7 +470,7 @@ public class JoinResultUpdater {
}
@Override
- public BindingSet next() {
+ public VisibilityBindingSet next() {
final ColumnIterator columns = rows.next().getValue();
while(columns.hasNext()) {
@@ -464,11 +478,7 @@ public class JoinResultUpdater {
final Entry<Column, Bytes> entry = columns.next();
if(BINDING_SET_COLUMNS.contains(entry.getKey())) {
final String bindingSetString = entry.getValue().toString();
- try {
- return converter.convert(bindingSetString, varOrder);
- } catch (BindingSetConversionException e) {
- throw new RuntimeException("Could not convert one of the stored BindingSets from a String: " + bindingSetString, e);
- }
+ return (VisibilityBindingSet) valueConverter.convert(bindingSetString, varOrder);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/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 f3ff089..8e0a6fe 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
@@ -26,7 +26,6 @@ import javax.annotation.ParametersAreNonnullByDefault;
import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryColumns;
import org.apache.rya.indexing.pcj.fluo.app.query.QueryMetadata;
import org.openrdf.query.Binding;
-import org.openrdf.query.BindingSet;
import org.openrdf.query.impl.MapBindingSet;
import io.fluo.api.client.TransactionBase;
@@ -34,8 +33,10 @@ import io.fluo.api.data.Bytes;
import io.fluo.api.data.Column;
import io.fluo.api.types.Encoder;
import io.fluo.api.types.StringEncoder;
+import mvm.rya.indexing.accumulo.VisibilityBindingSet;
import mvm.rya.indexing.external.tupleSet.BindingSetStringConverter;
import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
+import mvm.rya.indexing.external.tupleSet.VisibilityBindingSetStringConverter;
/**
* Updates the results of a Query node when one of its children has added a
@@ -43,10 +44,10 @@ import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
*/
@ParametersAreNonnullByDefault
public class QueryResultUpdater {
-
private final Encoder encoder = new StringEncoder();
private final BindingSetStringConverter converter = new BindingSetStringConverter();
+ private final VisibilityBindingSetStringConverter valueConverter = new VisibilityBindingSetStringConverter();
/**
* Updates the results of a Query node when one of its children has added a
@@ -58,7 +59,7 @@ public class QueryResultUpdater {
*/
public void updateQueryResults(
final TransactionBase tx,
- final BindingSet childBindingSet,
+ final VisibilityBindingSet childBindingSet,
final QueryMetadata queryMetadata) {
checkNotNull(tx);
checkNotNull(childBindingSet);
@@ -75,11 +76,12 @@ public class QueryResultUpdater {
}
}
final String queryBindingSetString = converter.convert(queryBindingSet, queryVarOrder);
+ final String queryBindingSetValueString = valueConverter.convert(new VisibilityBindingSet(queryBindingSet, childBindingSet.getVisibility()), queryVarOrder);
// Commit it to the Fluo table for the SPARQL query. This isn't guaranteed to be a new entry.
final Bytes row = encoder.encode(queryMetadata.getNodeId() + NODEID_BS_DELIM + queryBindingSetString);
final Column col = FluoQueryColumns.QUERY_BINDING_SET;
- final Bytes value = encoder.encode(queryBindingSetString);
+ final Bytes value = encoder.encode(queryBindingSetValueString);
tx.set(row, col, value);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/export/IncrementalResultExporter.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/export/IncrementalResultExporter.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/export/IncrementalResultExporter.java
index fbbae33..c2c031c 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/export/IncrementalResultExporter.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/export/IncrementalResultExporter.java
@@ -20,9 +20,8 @@ package org.apache.rya.indexing.pcj.fluo.app.export;
import javax.annotation.ParametersAreNonnullByDefault;
-import org.openrdf.query.BindingSet;
-
import io.fluo.api.types.TypedTransactionBase;
+import mvm.rya.indexing.accumulo.VisibilityBindingSet;
/**
* Exports a single Binding Set that is a new result for a SPARQL query to some
@@ -40,7 +39,7 @@ public interface IncrementalResultExporter {
* Fluo application. (not null)
* @throws ResultExportException The result could not be exported.
*/
- public void export(TypedTransactionBase tx, String queryId, BindingSet result) throws ResultExportException;
+ public void export(TypedTransactionBase tx, String queryId, VisibilityBindingSet result) throws ResultExportException;
/**
* A result could not be exported.
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/export/rya/RyaResultExporter.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/export/rya/RyaResultExporter.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/export/rya/RyaResultExporter.java
index 4d51798..5c8c719 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/export/rya/RyaResultExporter.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/export/rya/RyaResultExporter.java
@@ -25,10 +25,10 @@ import java.util.Collections;
import org.apache.accumulo.core.client.Connector;
import org.apache.rya.indexing.pcj.fluo.app.export.IncrementalResultExporter;
import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryColumns;
-import org.openrdf.query.BindingSet;
import io.fluo.api.data.Bytes;
import io.fluo.api.types.TypedTransactionBase;
+import mvm.rya.indexing.accumulo.VisibilityBindingSet;
import mvm.rya.indexing.external.tupleSet.PcjTables;
import mvm.rya.indexing.external.tupleSet.PcjTables.PcjException;
@@ -55,7 +55,7 @@ public class RyaResultExporter implements IncrementalResultExporter {
public void export(
final TypedTransactionBase fluoTx,
final String queryId,
- final BindingSet result) throws ResultExportException {
+ final VisibilityBindingSet result) throws ResultExportException {
checkNotNull(fluoTx);
checkNotNull(queryId);
checkNotNull(result);
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/BindingSetUpdater.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/BindingSetUpdater.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/BindingSetUpdater.java
index aa944e4..9bd0148 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/BindingSetUpdater.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/BindingSetUpdater.java
@@ -31,13 +31,15 @@ import org.apache.rya.indexing.pcj.fluo.app.query.FilterMetadata;
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.QueryMetadata;
-import org.openrdf.query.BindingSet;
import io.fluo.api.client.TransactionBase;
import io.fluo.api.data.Bytes;
import io.fluo.api.data.Column;
+import io.fluo.api.types.Encoder;
+import io.fluo.api.types.StringEncoder;
import io.fluo.api.types.TypedObserver;
import io.fluo.api.types.TypedTransactionBase;
+import mvm.rya.indexing.accumulo.VisibilityBindingSet;
import mvm.rya.indexing.external.tupleSet.BindingSetConverter.BindingSetConversionException;
/**
@@ -48,6 +50,7 @@ import mvm.rya.indexing.external.tupleSet.BindingSetConverter.BindingSetConversi
@ParametersAreNonnullByDefault
public abstract class BindingSetUpdater extends TypedObserver {
+ private final Encoder encoder = new StringEncoder();
// DAO
private final FluoQueryMetadataDAO queryDao = new FluoQueryMetadataDAO();
@@ -74,9 +77,10 @@ public abstract class BindingSetUpdater extends TypedObserver {
checkNotNull(row);
checkNotNull(col);
- final Observation observation = parseObservation( tx, BindingSetRow.make(row) );
+ final String bindingSetString = encoder.decodeString(tx.get(row, col));
+ final Observation observation = parseObservation( tx, new BindingSetRow(BindingSetRow.make(row).getNodeId(), bindingSetString) );
final String observedNodeId = observation.getObservedNodeId();
- final BindingSet observedBindingSet = observation.getObservedBindingSet();
+ final VisibilityBindingSet observedBindingSet = observation.getObservedBindingSet();
final String parentNodeId = observation.getParentId();
// Figure out which node needs to handle the new metadata.
@@ -100,7 +104,7 @@ public abstract class BindingSetUpdater extends TypedObserver {
final JoinMetadata parentJoin = queryDao.readJoinMetadata(tx, parentNodeId);
try {
joinUpdater.updateJoinResults(tx, observedNodeId, observedBindingSet, parentJoin);
- } catch (BindingSetConversionException e) {
+ } catch (final BindingSetConversionException e) {
throw new RuntimeException("Could not process a Join node.", e);
}
break;
@@ -117,7 +121,7 @@ public abstract class BindingSetUpdater extends TypedObserver {
public static final class Observation {
private final String observedNodeId;
- private final BindingSet observedBindingSet;
+ private final VisibilityBindingSet observedBindingSet;
private final String parentNodeId;
/**
@@ -129,7 +133,7 @@ public abstract class BindingSetUpdater extends TypedObserver {
*/
public Observation(
final String observedNodeId,
- final BindingSet observedBindingSet,
+ final VisibilityBindingSet observedBindingSet,
final String parentNodeId) {
this.observedNodeId = checkNotNull(observedNodeId);
this.observedBindingSet = checkNotNull(observedBindingSet);
@@ -146,7 +150,7 @@ public abstract class BindingSetUpdater extends TypedObserver {
/**
* @return A Binding Set that was just emitted.
*/
- public BindingSet getObservedBindingSet() {
+ public VisibilityBindingSet getObservedBindingSet() {
return observedBindingSet;
}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/FilterObserver.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/FilterObserver.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/FilterObserver.java
index 2accde3..fb15934 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/FilterObserver.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/FilterObserver.java
@@ -27,8 +27,9 @@ import org.apache.rya.indexing.pcj.fluo.app.query.FluoQueryMetadataDAO;
import org.openrdf.query.BindingSet;
import io.fluo.api.client.TransactionBase;
-import mvm.rya.indexing.external.tupleSet.BindingSetStringConverter;
+import mvm.rya.indexing.accumulo.VisibilityBindingSet;
import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
+import mvm.rya.indexing.external.tupleSet.VisibilityBindingSetStringConverter;
/**
* Notified when the results of a Filter have been updated to include a new
@@ -37,7 +38,7 @@ import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
*/
public class FilterObserver extends BindingSetUpdater {
- private final BindingSetStringConverter converter = new BindingSetStringConverter();
+ private final VisibilityBindingSetStringConverter converter = new VisibilityBindingSetStringConverter();
private final FluoQueryMetadataDAO queryDao = new FluoQueryMetadataDAO();
@@ -57,7 +58,7 @@ public class FilterObserver extends BindingSetUpdater {
// Read the Binding Set that was just emmitted by the Filter.
final VariableOrder filterVarOrder = filterMetadata.getVariableOrder();
- final BindingSet filterBindingSet = converter.convert(parsedRow.getBindingSetString(), filterVarOrder);
+ final VisibilityBindingSet filterBindingSet = (VisibilityBindingSet) converter.convert(parsedRow.getBindingSetString(), filterVarOrder);
// Figure out which node needs to handle the new metadata.
final String parentNodeId = filterMetadata.getParentNodeId();
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/JoinObserver.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/JoinObserver.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/JoinObserver.java
index 43b0a4e..a8cd0df 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/JoinObserver.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/JoinObserver.java
@@ -27,8 +27,9 @@ import org.apache.rya.indexing.pcj.fluo.app.query.JoinMetadata;
import org.openrdf.query.BindingSet;
import io.fluo.api.client.TransactionBase;
-import mvm.rya.indexing.external.tupleSet.BindingSetStringConverter;
+import mvm.rya.indexing.accumulo.VisibilityBindingSet;
import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
+import mvm.rya.indexing.external.tupleSet.VisibilityBindingSetStringConverter;
/**
* Notified when the results of a Join have been updated to include a new
@@ -37,7 +38,7 @@ import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
*/
public class JoinObserver extends BindingSetUpdater {
- private final BindingSetStringConverter converter = new BindingSetStringConverter();
+ private final VisibilityBindingSetStringConverter converter = new VisibilityBindingSetStringConverter();
private final FluoQueryMetadataDAO queryDao = new FluoQueryMetadataDAO();
@@ -56,7 +57,7 @@ public class JoinObserver extends BindingSetUpdater {
// Read the Binding Set that was just emmitted by the Join.
final VariableOrder joinVarOrder = joinMetadata.getVariableOrder();
- final BindingSet joinBindingSet = converter.convert(parsedRow.getBindingSetString(), joinVarOrder);
+ final VisibilityBindingSet joinBindingSet = (VisibilityBindingSet) converter.convert(parsedRow.getBindingSetString(), joinVarOrder);
// Figure out which node needs to handle the new metadata.
final String parentNodeId = joinMetadata.getParentNodeId();
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/QueryResultObserver.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/QueryResultObserver.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/QueryResultObserver.java
index 7c1a588..fe4dc56 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/QueryResultObserver.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/QueryResultObserver.java
@@ -29,17 +29,19 @@ import org.apache.rya.indexing.pcj.fluo.app.export.rya.RyaResultExporterFactory;
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.openrdf.query.BindingSet;
import com.google.common.base.Optional;
import com.google.common.collect.ImmutableSet;
import io.fluo.api.data.Bytes;
import io.fluo.api.data.Column;
+import io.fluo.api.types.Encoder;
+import io.fluo.api.types.StringEncoder;
import io.fluo.api.types.TypedObserver;
import io.fluo.api.types.TypedTransactionBase;
-import mvm.rya.indexing.external.tupleSet.BindingSetStringConverter;
+import mvm.rya.indexing.accumulo.VisibilityBindingSet;
import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
+import mvm.rya.indexing.external.tupleSet.VisibilityBindingSetStringConverter;
/**
* Performs incremental result exporting to the configured destinations.
@@ -47,9 +49,9 @@ import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
public class QueryResultObserver extends TypedObserver {
private static final Logger log = Logger.getLogger(QueryResultObserver.class);
- private final FluoQueryMetadataDAO queryDao = new FluoQueryMetadataDAO();
-
- private final BindingSetStringConverter converter = new BindingSetStringConverter();
+ private static final FluoQueryMetadataDAO QUERY_DAO = new FluoQueryMetadataDAO();
+ private static final Encoder ENCODER = new StringEncoder();
+ private static final VisibilityBindingSetStringConverter CONVERTER = new VisibilityBindingSetStringConverter();
/**
* Builders for each type of result exporter we support.
@@ -93,16 +95,16 @@ public class QueryResultObserver extends TypedObserver {
@Override
public void process(final TypedTransactionBase tx, final Bytes row, final Column col) {
// Read the SPARQL query and it Binding Set from the row id.
- final String[] queryAndBindingSet = row.toString().split(NODEID_BS_DELIM);
+ final String[] queryAndBindingSet = ENCODER.decodeString(row).split(NODEID_BS_DELIM);
final String queryId = queryAndBindingSet[0];
- final String bindingSetString = queryAndBindingSet[1];
+ final String bindingSetString = ENCODER.decodeString(tx.get(row, col));
// Fetch the query's Variable Order from the Fluo table.
- final QueryMetadata queryMetadata = queryDao.readQueryMetadata(tx, queryId);
+ final QueryMetadata queryMetadata = QUERY_DAO.readQueryMetadata(tx, queryId);
final VariableOrder varOrder = queryMetadata.getVariableOrder();
// Export the result using each of the provided exporters.
- BindingSet result = converter.convert(bindingSetString, varOrder);
+ final VisibilityBindingSet result = (VisibilityBindingSet) CONVERTER.convert(bindingSetString, varOrder);
for(final IncrementalResultExporter exporter : exporters) {
try {
exporter.export(tx, queryId, result);
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/StatementPatternObserver.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/StatementPatternObserver.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/StatementPatternObserver.java
index ddba9a2..7b1e510 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/StatementPatternObserver.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/StatementPatternObserver.java
@@ -27,8 +27,9 @@ import org.apache.rya.indexing.pcj.fluo.app.query.StatementPatternMetadata;
import org.openrdf.query.BindingSet;
import io.fluo.api.client.TransactionBase;
-import mvm.rya.indexing.external.tupleSet.BindingSetStringConverter;
+import mvm.rya.indexing.accumulo.VisibilityBindingSet;
import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
+import mvm.rya.indexing.external.tupleSet.VisibilityBindingSetStringConverter;
/**
* Notified when the results of a Statement Pattern have been updated to include
@@ -37,7 +38,7 @@ import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
*/
public class StatementPatternObserver extends BindingSetUpdater {
- private final BindingSetStringConverter converter = new BindingSetStringConverter();
+ private static final VisibilityBindingSetStringConverter CONVERTER = new VisibilityBindingSetStringConverter();
// DAO
private final FluoQueryMetadataDAO queryDao = new FluoQueryMetadataDAO();
@@ -54,10 +55,11 @@ public class StatementPatternObserver extends BindingSetUpdater {
// Read the Statement Pattern metadata.
final String spNodeId = parsedRow.getNodeId();
final StatementPatternMetadata spMetadata = queryDao.readStatementPatternMetadata(tx, spNodeId);
+ final String bindingSetValue = parsedRow.getBindingSetString();
// Read the Binding Set that was just emmitted by the Statement Pattern.
final VariableOrder spVarOrder = spMetadata.getVariableOrder();
- final BindingSet spBindingSet = converter.convert(parsedRow.getBindingSetString(), spVarOrder);
+ final VisibilityBindingSet spBindingSet = (VisibilityBindingSet) CONVERTER.convert(bindingSetValue, spVarOrder);
// Figure out which node needs to handle the new metadata.
final String parentNodeId = spMetadata.getParentNodeId();
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/TripleObserver.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/TripleObserver.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/TripleObserver.java
index d43ffc9..496c0ed 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/TripleObserver.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/observers/TripleObserver.java
@@ -40,8 +40,13 @@ import io.fluo.api.data.Column;
import io.fluo.api.data.Span;
import io.fluo.api.iterator.ColumnIterator;
import io.fluo.api.iterator.RowIterator;
+import io.fluo.api.types.Encoder;
+import io.fluo.api.types.StringEncoder;
import io.fluo.api.types.TypedObserver;
import io.fluo.api.types.TypedTransactionBase;
+import mvm.rya.indexing.accumulo.VisibilityBindingSet;
+import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
+import mvm.rya.indexing.external.tupleSet.VisibilityBindingSetStringConverter;
/**
* An observer that matches new Triples to the Statement Patterns that are part
@@ -50,7 +55,9 @@ import io.fluo.api.types.TypedTransactionBase;
*/
public class TripleObserver extends TypedObserver {
- private final FluoQueryMetadataDAO queryDao = new FluoQueryMetadataDAO();
+ private static final Encoder ENCODER = new StringEncoder();
+ private static final FluoQueryMetadataDAO QUERY_DAO = new FluoQueryMetadataDAO();
+ private static final VisibilityBindingSetStringConverter CONVERTER = new VisibilityBindingSetStringConverter();
public TripleObserver() {
super(new StringTypeLayer());
@@ -65,6 +72,11 @@ public class TripleObserver extends TypedObserver {
public void process(final TypedTransactionBase tx, final Bytes row, final Column column) {
//get string representation of triple
final String triple = IncUpdateDAO.getTripleString(row);
+ final Bytes visiBytes = tx.get(row, FluoQueryColumns.TRIPLES);
+ String visibility = "";
+ if(visiBytes != null) {
+ visibility = ENCODER.decodeString(visiBytes);
+ }
//get variable metadata for all SP in table
final ScannerConfiguration sc1 = new ScannerConfiguration();
@@ -75,19 +87,25 @@ public class TripleObserver extends TypedObserver {
final RowIterator ri = tx.get(sc1);
while(ri.hasNext()) {
-
final Entry<Bytes, ColumnIterator> next = ri.next();
final ColumnIterator ci = next.getValue();
final String spID = next.getKey().toString();
- final StatementPatternMetadata spMetadata = queryDao.readStatementPatternMetadata(tx, spID);
+ final StatementPatternMetadata spMetadata = QUERY_DAO.readStatementPatternMetadata(tx, spID);
final String pattern = spMetadata.getStatementPattern();
while(ci.hasNext()) {
final String varOrders = ci.next().getValue().toString();
- final String bindingSet = getBindingSet(triple, pattern, varOrders);
- if(bindingSet.length() != 0) {
- tx.mutate().row(spID + NODEID_BS_DELIM + bindingSet).col(FluoQueryColumns.STATEMENT_PATTERN_BINDING_SET).set(bindingSet);
+ final VariableOrder varOrder = new VariableOrder(varOrders);
+ final String bindingSetString = getBindingSet(triple, pattern, varOrders);
+
+ //Statement matches to a binding set
+ if(bindingSetString.length() != 0) {
+ final VisibilityBindingSet bindingSet = new VisibilityBindingSet(
+ CONVERTER.convert(bindingSetString, varOrder),
+ visibility);
+ final String valueString = CONVERTER.convert(bindingSet, varOrder);
+ tx.mutate().row(spID + NODEID_BS_DELIM + bindingSetString).col(FluoQueryColumns.STATEMENT_PATTERN_BINDING_SET).set(valueString);
}
}
}
@@ -96,8 +114,16 @@ public class TripleObserver extends TypedObserver {
tx.delete(row, column);
}
- //determines whether triple matches SPID conditions and generates bindingset
- //whose order is determined by varOrder
+ /**
+ * Determines whether triple matches Statement Pattern ID conditions if
+ * so, generates a string representation of a BindingSet whose order
+ * is determined by varOrder.
+ * @param triple - The triple to consider.
+ * @param spID - The statement pattern ID
+ * @param varOrder - The variable order
+ * @return The string representation of the BindingSet or an empty string,
+ * signifying the triple did not match the statement pattern ID.
+ */
private static String getBindingSet(final String triple, final String spID, final String varOrder) {
final String[] spIdArray = spID.split(DELIM);
final String[] tripleArray = triple.split(DELIM);
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryColumns.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryColumns.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryColumns.java
index fa25456..be24ac9 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryColumns.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryColumns.java
@@ -88,7 +88,7 @@ public class FluoQueryColumns {
* <p>
* <table border="1" style="width:100%">
* <tr> <th>Fluo Row</td> <th>Fluo Column</td> <th>Fluo Value</td> </tr>
- * <tr> <td>Core Rya SPO formatted triple</td> <td>triples:SPO</td> <td>empty</td> </tr>
+ * <tr> <td>Core Rya SPO formatted triple</td> <td>triples:SPO</td> <td>visibility</td> </tr>
* </table>
*/
public static final Column TRIPLES = new Column("triples", "SPO");
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryMetadataDAO.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryMetadataDAO.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryMetadataDAO.java
index 265ca0f..83985da 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryMetadataDAO.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/main/java/org/apache/rya/indexing/pcj/fluo/app/query/FluoQueryMetadataDAO.java
@@ -205,8 +205,8 @@ public class FluoQueryMetadataDAO {
final String varOrderString = encoder.decodeString( values.get(FluoQueryColumns.JOIN_VARIABLE_ORDER));
final VariableOrder varOrder = new VariableOrder(varOrderString);
- String joinTypeString = encoder.decodeString( values.get(FluoQueryColumns.JOIN_TYPE) );
- JoinType joinType = JoinType.valueOf(joinTypeString);
+ final String joinTypeString = encoder.decodeString( values.get(FluoQueryColumns.JOIN_TYPE) );
+ final JoinType joinType = JoinType.valueOf(joinTypeString);
final String parentNodeId = encoder.decodeString( values.get(FluoQueryColumns.JOIN_PARENT_NODE_ID) );
final String leftChildNodeId = encoder.decodeString( values.get(FluoQueryColumns.JOIN_LEFT_CHILD_NODE_ID) );
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.app/src/test/java/org/apache/rya/indexing/pcj/fluo/app/LeftOuterJoinTest.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/test/java/org/apache/rya/indexing/pcj/fluo/app/LeftOuterJoinTest.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/test/java/org/apache/rya/indexing/pcj/fluo/app/LeftOuterJoinTest.java
index 025c3e7..08d5cef 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/test/java/org/apache/rya/indexing/pcj/fluo/app/LeftOuterJoinTest.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/test/java/org/apache/rya/indexing/pcj/fluo/app/LeftOuterJoinTest.java
@@ -37,6 +37,8 @@ import org.openrdf.query.impl.MapBindingSet;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
+import mvm.rya.indexing.accumulo.VisibilityBindingSet;
+
/**
* Tests the methods of {@link LeftOuterJoin}.
*/
@@ -46,128 +48,134 @@ public class LeftOuterJoinTest {
@Test
public void newLeftResult_noRightMatches() {
- IterativeJoin leftOuterJoin = new LeftOuterJoin();
+ final IterativeJoin leftOuterJoin = new LeftOuterJoin();
// There is a new left result.
- MapBindingSet newLeftResult = new MapBindingSet();
- newLeftResult.addBinding("name", vf.createLiteral("Bob"));
+ final MapBindingSet mapLeftResult = new MapBindingSet();
+ mapLeftResult.addBinding("name", vf.createLiteral("Bob"));
+ final VisibilityBindingSet newLeftResult = new VisibilityBindingSet(mapLeftResult);
// There are no right results that join with the left result.
- Iterator<BindingSet> rightResults= new ArrayList<BindingSet>().iterator();
+ final Iterator<VisibilityBindingSet> rightResults= new ArrayList<VisibilityBindingSet>().iterator();
// Therefore, the left result is a new join result.
- Iterator<BindingSet> newJoinResultsIt = leftOuterJoin.newLeftResult(newLeftResult, rightResults);
+ final Iterator<VisibilityBindingSet> newJoinResultsIt = leftOuterJoin.newLeftResult(newLeftResult, rightResults);
- Set<BindingSet> newJoinResults = new HashSet<>();
+ final Set<BindingSet> newJoinResults = new HashSet<>();
while(newJoinResultsIt.hasNext()) {
newJoinResults.add( newJoinResultsIt.next() );
}
- Set<BindingSet> expected = Sets.<BindingSet>newHashSet( newLeftResult );
+ final Set<BindingSet> expected = Sets.<BindingSet>newHashSet( newLeftResult );
assertEquals(expected, newJoinResults);
}
@Test
public void newLeftResult_joinsWithRightResults() {
- IterativeJoin leftOuterJoin = new LeftOuterJoin();
+ final IterativeJoin leftOuterJoin = new LeftOuterJoin();
// There is a new left result.
- MapBindingSet newLeftResult = new MapBindingSet();
- newLeftResult.addBinding("name", vf.createLiteral("Bob"));
- newLeftResult.addBinding("height", vf.createLiteral("5'9\""));
+ final MapBindingSet mapLeftResult = new MapBindingSet();
+ mapLeftResult.addBinding("name", vf.createLiteral("Bob"));
+ mapLeftResult.addBinding("height", vf.createLiteral("5'9\""));
+ final VisibilityBindingSet newLeftResult = new VisibilityBindingSet(mapLeftResult);
// There are a few right results that join with the left result.
- MapBindingSet nameAge = new MapBindingSet();
+ final MapBindingSet nameAge = new MapBindingSet();
nameAge.addBinding("name", vf.createLiteral("Bob"));
nameAge.addBinding("age", vf.createLiteral(56));
+ final VisibilityBindingSet visiAge = new VisibilityBindingSet(nameAge);
- MapBindingSet nameHair = new MapBindingSet();
+ final MapBindingSet nameHair = new MapBindingSet();
nameHair.addBinding("name", vf.createLiteral("Bob"));
nameHair.addBinding("hairColor", vf.createLiteral("Brown"));
+ final VisibilityBindingSet visiHair = new VisibilityBindingSet(nameHair);
- Iterator<BindingSet> rightResults = Lists.<BindingSet>newArrayList(nameAge, nameHair).iterator();
+ final Iterator<VisibilityBindingSet> rightResults = Lists.<VisibilityBindingSet>newArrayList(visiAge, visiHair).iterator();
// Therefore, there are a few new join results that mix the two together.
- Iterator<BindingSet> newJoinResultsIt = leftOuterJoin.newLeftResult(newLeftResult, rightResults);
+ final Iterator<VisibilityBindingSet> newJoinResultsIt = leftOuterJoin.newLeftResult(newLeftResult, rightResults);
- Set<BindingSet> newJoinResults = new HashSet<>();
+ final Set<BindingSet> newJoinResults = new HashSet<>();
while(newJoinResultsIt.hasNext()) {
newJoinResults.add( newJoinResultsIt.next() );
}
- Set<BindingSet> expected = Sets.<BindingSet>newHashSet();
- MapBindingSet nameHeightAge = new MapBindingSet();
+ final Set<BindingSet> expected = Sets.<BindingSet>newHashSet();
+ final MapBindingSet nameHeightAge = new MapBindingSet();
nameHeightAge.addBinding("name", vf.createLiteral("Bob"));
nameHeightAge.addBinding("height", vf.createLiteral("5'9\""));
nameHeightAge.addBinding("age", vf.createLiteral(56));
- expected.add(nameHeightAge);
+ expected.add(new VisibilityBindingSet(nameHeightAge));
- MapBindingSet nameHeightHair = new MapBindingSet();
+ final MapBindingSet nameHeightHair = new MapBindingSet();
nameHeightHair.addBinding("name", vf.createLiteral("Bob"));
nameHeightHair.addBinding("height", vf.createLiteral("5'9\""));
nameHeightHair.addBinding("hairColor", vf.createLiteral("Brown"));
- expected.add(nameHeightHair);
+ expected.add(new VisibilityBindingSet(nameHeightHair));
assertEquals(expected, newJoinResults);
}
@Test
public void newRightResult_noLeftMatches() {
- IterativeJoin leftOuterJoin = new LeftOuterJoin();
+ final IterativeJoin leftOuterJoin = new LeftOuterJoin();
// There are no left results.
- Iterator<BindingSet> leftResults= new ArrayList<BindingSet>().iterator();
+ final Iterator<VisibilityBindingSet> leftResults= new ArrayList<VisibilityBindingSet>().iterator();
// There is a new right result.
- MapBindingSet newRightResult = new MapBindingSet();
+ final MapBindingSet newRightResult = new MapBindingSet();
newRightResult.addBinding("name", vf.createLiteral("Bob"));
// Therefore, there are no new join results.
- Iterator<BindingSet> newJoinResultsIt = leftOuterJoin.newRightResult(leftResults, newRightResult);
+ final Iterator<VisibilityBindingSet> newJoinResultsIt = leftOuterJoin.newRightResult(leftResults, new VisibilityBindingSet(newRightResult));
assertFalse( newJoinResultsIt.hasNext() );
}
@Test
public void newRightResult_joinsWithLeftResults() {
- IterativeJoin leftOuterJoin = new LeftOuterJoin();
+ final IterativeJoin leftOuterJoin = new LeftOuterJoin();
// There are a few left results that join with the new right result.
- MapBindingSet nameAge = new MapBindingSet();
+ final MapBindingSet nameAge = new MapBindingSet();
nameAge.addBinding("name", vf.createLiteral("Bob"));
nameAge.addBinding("age", vf.createLiteral(56));
- MapBindingSet nameHair = new MapBindingSet();
+ final MapBindingSet nameHair = new MapBindingSet();
nameHair.addBinding("name", vf.createLiteral("Bob"));
nameHair.addBinding("hairColor", vf.createLiteral("Brown"));
- Iterator<BindingSet> leftResults = Lists.<BindingSet>newArrayList(nameAge, nameHair).iterator();
+ final Iterator<VisibilityBindingSet> leftResults = Lists.<VisibilityBindingSet>newArrayList(
+ new VisibilityBindingSet(nameAge),
+ new VisibilityBindingSet(nameHair)).iterator();
// There is a new right result.
- MapBindingSet newRightResult = new MapBindingSet();
+ final MapBindingSet newRightResult = new MapBindingSet();
newRightResult.addBinding("name", vf.createLiteral("Bob"));
newRightResult.addBinding("height", vf.createLiteral("5'9\""));
// Therefore, there are a few new join results that mix the two together.
- Iterator<BindingSet> newJoinResultsIt = leftOuterJoin.newRightResult(leftResults, newRightResult);
+ final Iterator<VisibilityBindingSet> newJoinResultsIt = leftOuterJoin.newRightResult(leftResults, new VisibilityBindingSet(newRightResult));
- Set<BindingSet> newJoinResults = new HashSet<>();
+ final Set<BindingSet> newJoinResults = new HashSet<>();
while(newJoinResultsIt.hasNext()) {
newJoinResults.add( newJoinResultsIt.next() );
}
- Set<BindingSet> expected = Sets.<BindingSet>newHashSet();
- MapBindingSet nameHeightAge = new MapBindingSet();
+ final Set<BindingSet> expected = Sets.<BindingSet>newHashSet();
+ final MapBindingSet nameHeightAge = new MapBindingSet();
nameHeightAge.addBinding("name", vf.createLiteral("Bob"));
nameHeightAge.addBinding("height", vf.createLiteral("5'9\""));
nameHeightAge.addBinding("age", vf.createLiteral(56));
- expected.add(nameHeightAge);
+ expected.add(new VisibilityBindingSet(nameHeightAge));
- MapBindingSet nameHeightHair = new MapBindingSet();
+ final MapBindingSet nameHeightHair = new MapBindingSet();
nameHeightHair.addBinding("name", vf.createLiteral("Bob"));
nameHeightHair.addBinding("height", vf.createLiteral("5'9\""));
nameHeightHair.addBinding("hairColor", vf.createLiteral("Brown"));
- expected.add(nameHeightHair);
+ expected.add(new VisibilityBindingSet(nameHeightHair));
assertEquals(expected, newJoinResults);
}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.app/src/test/java/org/apache/rya/indexing/pcj/fluo/app/NaturalJoinTest.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.app/src/test/java/org/apache/rya/indexing/pcj/fluo/app/NaturalJoinTest.java b/extras/rya.pcj.fluo/pcj.fluo.app/src/test/java/org/apache/rya/indexing/pcj/fluo/app/NaturalJoinTest.java
index 15023c5..651ea11 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.app/src/test/java/org/apache/rya/indexing/pcj/fluo/app/NaturalJoinTest.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.app/src/test/java/org/apache/rya/indexing/pcj/fluo/app/NaturalJoinTest.java
@@ -37,6 +37,8 @@ import org.openrdf.query.impl.MapBindingSet;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
+import mvm.rya.indexing.accumulo.VisibilityBindingSet;
+
/**
* Tests the methods of {@link NaturalJoin}.
*/
@@ -46,120 +48,124 @@ public class NaturalJoinTest {
@Test
public void newLeftResult_noRightMatches() {
- IterativeJoin naturalJoin = new NaturalJoin();
+ final IterativeJoin naturalJoin = new NaturalJoin();
// There is a new left result.
- MapBindingSet newLeftResult = new MapBindingSet();
+ final MapBindingSet newLeftResult = new MapBindingSet();
newLeftResult.addBinding("name", vf.createLiteral("Bob"));
// There are no right results that join with the left result.
- Iterator<BindingSet> rightResults= new ArrayList<BindingSet>().iterator();
+ final Iterator<VisibilityBindingSet> rightResults= new ArrayList<VisibilityBindingSet>().iterator();
// Therefore, the left result is a new join result.
- Iterator<BindingSet> newJoinResultsIt = naturalJoin.newLeftResult(newLeftResult, rightResults);
+ final Iterator<VisibilityBindingSet> newJoinResultsIt = naturalJoin.newLeftResult(new VisibilityBindingSet(newLeftResult), rightResults);
assertFalse( newJoinResultsIt.hasNext() );
}
@Test
public void newLeftResult_joinsWithRightResults() {
- IterativeJoin naturalJoin = new NaturalJoin();
+ final IterativeJoin naturalJoin = new NaturalJoin();
// There is a new left result.
- MapBindingSet newLeftResult = new MapBindingSet();
+ final MapBindingSet newLeftResult = new MapBindingSet();
newLeftResult.addBinding("name", vf.createLiteral("Bob"));
newLeftResult.addBinding("height", vf.createLiteral("5'9\""));
// There are a few right results that join with the left result.
- MapBindingSet nameAge = new MapBindingSet();
+ final MapBindingSet nameAge = new MapBindingSet();
nameAge.addBinding("name", vf.createLiteral("Bob"));
nameAge.addBinding("age", vf.createLiteral(56));
- MapBindingSet nameHair = new MapBindingSet();
+ final MapBindingSet nameHair = new MapBindingSet();
nameHair.addBinding("name", vf.createLiteral("Bob"));
nameHair.addBinding("hairColor", vf.createLiteral("Brown"));
- Iterator<BindingSet> rightResults = Lists.<BindingSet>newArrayList(nameAge, nameHair).iterator();
+ final Iterator<VisibilityBindingSet> rightResults = Lists.<VisibilityBindingSet>newArrayList(
+ new VisibilityBindingSet(nameAge),
+ new VisibilityBindingSet(nameHair)).iterator();
// Therefore, there are a few new join results that mix the two together.
- Iterator<BindingSet> newJoinResultsIt = naturalJoin.newLeftResult(newLeftResult, rightResults);
+ final Iterator<VisibilityBindingSet> newJoinResultsIt = naturalJoin.newLeftResult(new VisibilityBindingSet(newLeftResult), rightResults);
- Set<BindingSet> newJoinResults = new HashSet<>();
+ final Set<BindingSet> newJoinResults = new HashSet<>();
while(newJoinResultsIt.hasNext()) {
newJoinResults.add( newJoinResultsIt.next() );
}
- Set<BindingSet> expected = Sets.<BindingSet>newHashSet();
- MapBindingSet nameHeightAge = new MapBindingSet();
+ final Set<BindingSet> expected = Sets.<BindingSet>newHashSet();
+ final MapBindingSet nameHeightAge = new MapBindingSet();
nameHeightAge.addBinding("name", vf.createLiteral("Bob"));
nameHeightAge.addBinding("height", vf.createLiteral("5'9\""));
nameHeightAge.addBinding("age", vf.createLiteral(56));
- expected.add(nameHeightAge);
+ expected.add(new VisibilityBindingSet(nameHeightAge));
- MapBindingSet nameHeightHair = new MapBindingSet();
+ final MapBindingSet nameHeightHair = new MapBindingSet();
nameHeightHair.addBinding("name", vf.createLiteral("Bob"));
nameHeightHair.addBinding("height", vf.createLiteral("5'9\""));
nameHeightHair.addBinding("hairColor", vf.createLiteral("Brown"));
- expected.add(nameHeightHair);
+ expected.add(new VisibilityBindingSet(nameHeightHair));
assertEquals(expected, newJoinResults);
}
@Test
public void newRightResult_noLeftMatches() {
- IterativeJoin naturalJoin = new NaturalJoin();
+ final IterativeJoin naturalJoin = new NaturalJoin();
// There are no left results.
- Iterator<BindingSet> leftResults= new ArrayList<BindingSet>().iterator();
+ final Iterator<VisibilityBindingSet> leftResults= new ArrayList<VisibilityBindingSet>().iterator();
// There is a new right result.
- MapBindingSet newRightResult = new MapBindingSet();
+ final MapBindingSet newRightResult = new MapBindingSet();
newRightResult.addBinding("name", vf.createLiteral("Bob"));
// Therefore, there are no new join results.
- Iterator<BindingSet> newJoinResultsIt = naturalJoin.newRightResult(leftResults, newRightResult);
+ final Iterator<VisibilityBindingSet> newJoinResultsIt = naturalJoin.newRightResult(leftResults, new VisibilityBindingSet(newRightResult));
assertFalse( newJoinResultsIt.hasNext() );
}
@Test
public void newRightResult_joinsWithLeftResults() {
- IterativeJoin naturalJoin = new NaturalJoin();
+ final IterativeJoin naturalJoin = new NaturalJoin();
// There are a few left results that join with the new right result.
- MapBindingSet nameAge = new MapBindingSet();
+ final MapBindingSet nameAge = new MapBindingSet();
nameAge.addBinding("name", vf.createLiteral("Bob"));
nameAge.addBinding("age", vf.createLiteral(56));
- MapBindingSet nameHair = new MapBindingSet();
+ final MapBindingSet nameHair = new MapBindingSet();
nameHair.addBinding("name", vf.createLiteral("Bob"));
nameHair.addBinding("hairColor", vf.createLiteral("Brown"));
- Iterator<BindingSet> leftResults = Lists.<BindingSet>newArrayList(nameAge, nameHair).iterator();
+ final Iterator<VisibilityBindingSet> leftResults = Lists.<VisibilityBindingSet>newArrayList(
+ new VisibilityBindingSet(nameAge),
+ new VisibilityBindingSet(nameHair)).iterator();
// There is a new right result.
- MapBindingSet newRightResult = new MapBindingSet();
+ final MapBindingSet newRightResult = new MapBindingSet();
newRightResult.addBinding("name", vf.createLiteral("Bob"));
newRightResult.addBinding("height", vf.createLiteral("5'9\""));
// Therefore, there are a few new join results that mix the two together.
- Iterator<BindingSet> newJoinResultsIt = naturalJoin.newRightResult(leftResults, newRightResult);
+ final Iterator<VisibilityBindingSet> newJoinResultsIt = naturalJoin.newRightResult(leftResults, new VisibilityBindingSet(newRightResult));
- Set<BindingSet> newJoinResults = new HashSet<>();
+ final Set<BindingSet> newJoinResults = new HashSet<>();
while(newJoinResultsIt.hasNext()) {
newJoinResults.add( newJoinResultsIt.next() );
}
- Set<BindingSet> expected = Sets.<BindingSet>newHashSet();
- MapBindingSet nameHeightAge = new MapBindingSet();
+ final Set<BindingSet> expected = Sets.<BindingSet>newHashSet();
+ final MapBindingSet nameHeightAge = new MapBindingSet();
nameHeightAge.addBinding("name", vf.createLiteral("Bob"));
nameHeightAge.addBinding("height", vf.createLiteral("5'9\""));
nameHeightAge.addBinding("age", vf.createLiteral(56));
- expected.add(nameHeightAge);
+ expected.add(new VisibilityBindingSet(nameHeightAge));
- MapBindingSet nameHeightHair = new MapBindingSet();
+ final MapBindingSet nameHeightHair = new MapBindingSet();
nameHeightHair.addBinding("name", vf.createLiteral("Bob"));
nameHeightHair.addBinding("height", vf.createLiteral("5'9\""));
nameHeightHair.addBinding("hairColor", vf.createLiteral("Brown"));
- expected.add(nameHeightHair);
+ expected.add(new VisibilityBindingSet(nameHeightHair));
assertEquals(expected, newJoinResults);
}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/FluoLoader.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/FluoLoader.java b/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/FluoLoader.java
index 3168a71..7be539a 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/FluoLoader.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.client/src/main/java/org/apache/rya/indexing/pcj/fluo/client/util/FluoLoader.java
@@ -30,6 +30,8 @@ import org.openrdf.rio.RDFHandlerException;
import org.openrdf.rio.RDFParser;
import org.openrdf.rio.helpers.RDFHandlerBase;
+import com.google.common.base.Optional;
+
import io.fluo.api.client.FluoClient;
import mvm.rya.api.domain.RyaStatement;
import mvm.rya.api.resolver.RdfToRyaConversions;
@@ -68,7 +70,7 @@ public class FluoLoader extends RDFHandlerBase {
// If the buffer is full, flush it to the Fluo table.
if(buff.size() == FLUSH_SIZE) {
log.trace("Flushing " + buff.size() + " Statements from the buffer to Fluo.");
- insertTriples.insert(fluoClient, buff);
+ insertTriples.insert(fluoClient, buff, Optional.<String>absent());
buff.clear();
}
@@ -83,7 +85,7 @@ public class FluoLoader extends RDFHandlerBase {
if(!buff.isEmpty()) {
log.trace("Flushing the last " + buff.size() + " Statements from the buffer to Fluo.");
- insertTriples.insert(fluoClient, buff);
+ insertTriples.insert(fluoClient, buff, Optional.<String>absent());
buff.clear();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.demo/src/main/java/org/apache/rya/indexing/pcj/fluo/demo/FluoAndHistoricPcjsDemo.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.demo/src/main/java/org/apache/rya/indexing/pcj/fluo/demo/FluoAndHistoricPcjsDemo.java b/extras/rya.pcj.fluo/pcj.fluo.demo/src/main/java/org/apache/rya/indexing/pcj/fluo/demo/FluoAndHistoricPcjsDemo.java
index 0cbfa9a..eb7fb17 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.demo/src/main/java/org/apache/rya/indexing/pcj/fluo/demo/FluoAndHistoricPcjsDemo.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.demo/src/main/java/org/apache/rya/indexing/pcj/fluo/demo/FluoAndHistoricPcjsDemo.java
@@ -48,6 +48,7 @@ import org.openrdf.repository.RepositoryConnection;
import org.openrdf.repository.RepositoryException;
import org.openrdf.sail.SailException;
+import com.google.common.base.Optional;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
@@ -60,7 +61,6 @@ import mvm.rya.api.domain.RyaStatement;
import mvm.rya.api.domain.RyaType;
import mvm.rya.api.domain.RyaURI;
import mvm.rya.api.resolver.RyaToRdfConversions;
-import mvm.rya.api.resolver.RyaTypeResolverException;
import mvm.rya.indexing.external.tupleSet.AccumuloPcjSerializer;
import mvm.rya.indexing.external.tupleSet.BindingSetConverter.BindingSetConversionException;
import mvm.rya.indexing.external.tupleSet.PcjTables;
@@ -77,7 +77,7 @@ public class FluoAndHistoricPcjsDemo implements Demo {
private static final Logger log = Logger.getLogger(FluoAndHistoricPcjsDemo.class);
private static final AccumuloPcjSerializer converter = new AccumuloPcjSerializer();
-
+
// Employees
private static final RyaURI alice = new RyaURI("http://Alice");
private static final RyaURI bob = new RyaURI("http://Bob");
@@ -290,7 +290,7 @@ public class FluoAndHistoricPcjsDemo implements Demo {
private static void loadDataIntoFluo(final FluoClient fluoClient, final Set<RyaStatement> statements) {
final InsertTriples insertTriples = new InsertTriples();
for(final RyaStatement statement : statements) {
- insertTriples.insert(fluoClient, statement);
+ insertTriples.insert(fluoClient, statement, Optional.<String>absent());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/api/CountStatementsIT.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/api/CountStatementsIT.java b/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/api/CountStatementsIT.java
index 41c4f08..566d2d2 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/api/CountStatementsIT.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/api/CountStatementsIT.java
@@ -28,6 +28,7 @@ import java.util.List;
import org.apache.rya.indexing.pcj.fluo.ITBase;
import org.junit.Test;
+import com.google.common.base.Optional;
import com.google.common.io.Files;
import io.fluo.api.client.FluoFactory;
@@ -75,7 +76,7 @@ public class CountStatementsIT extends ITBase {
triples.add( RyaStatement.builder().setSubject(new RyaURI("http://David")).setPredicate(new RyaURI("http://talksTo")).setObject(new RyaURI("http://Bob")).build() );
triples.add( RyaStatement.builder().setSubject(new RyaURI("http://Eve")).setPredicate(new RyaURI("http://talksTo")).setObject(new RyaURI("http://Bob")).build() );
- new InsertTriples().insert(fluoClient, triples);
+ new InsertTriples().insert(fluoClient, triples, Optional.<String>absent());
// Load some statements into the Fluo app.
final BigInteger count = new CountStatements().countStatements(fluoClient);
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/api/GetQueryReportIT.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/api/GetQueryReportIT.java b/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/api/GetQueryReportIT.java
index 157412a..0e766b1 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/api/GetQueryReportIT.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/api/GetQueryReportIT.java
@@ -33,6 +33,7 @@ import org.apache.rya.indexing.pcj.fluo.app.query.FluoQuery;
import org.apache.rya.indexing.pcj.fluo.app.query.StatementPatternMetadata;
import org.junit.Test;
+import com.google.common.base.Optional;
import com.google.common.collect.Sets;
import mvm.rya.api.domain.RyaStatement;
@@ -77,7 +78,7 @@ public class GetQueryReportIT extends ITBase {
new CreatePcj().withRyaIntegration(fluoClient, RYA_TABLE_PREFIX, ryaRepo, accumuloConn, new HashSet<VariableOrder>(), sparql);
// Stream the data into Fluo.
- new InsertTriples().insert(fluoClient, streamedTriples);
+ new InsertTriples().insert(fluoClient, streamedTriples, Optional.<String>absent());
// Wait for the results to finish processing.
fluo.waitForObservers();
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/integration/InputIT.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/integration/InputIT.java b/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/integration/InputIT.java
index 41c2d7d..6e633f6 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/integration/InputIT.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/integration/InputIT.java
@@ -33,6 +33,7 @@ import org.openrdf.model.impl.URIImpl;
import org.openrdf.query.BindingSet;
import org.openrdf.query.impl.BindingImpl;
+import com.google.common.base.Optional;
import com.google.common.collect.Sets;
import mvm.rya.api.domain.RyaStatement;
@@ -135,7 +136,7 @@ public class InputIT extends ITBase {
assertTrue( results.isEmpty() );
// Stream the data into Fluo.
- new InsertTriples().insert(fluoClient, streamedTriples);
+ new InsertTriples().insert(fluoClient, streamedTriples, Optional.<String>absent());
// Verify the end results of the query match the expected results.
fluo.waitForObservers();
@@ -187,7 +188,7 @@ public class InputIT extends ITBase {
assertEquals(expected, results);
// Stream the data into Fluo.
- new InsertTriples().insert(fluoClient, streamedTriples);
+ new InsertTriples().insert(fluoClient, streamedTriples, Optional.<String>absent());
// Verify the end results of the query also include Frank.
fluo.waitForObservers();
@@ -244,7 +245,7 @@ public class InputIT extends ITBase {
assertEquals(expected, results);
// Stream the same Alice triple into Fluo.
- new InsertTriples().insert(fluoClient, streamedTriples);
+ new InsertTriples().insert(fluoClient, streamedTriples, Optional.<String>absent());
// Verify the end results of the query is stiill only Alice.
fluo.waitForObservers();
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/integration/QueryIT.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/integration/QueryIT.java b/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/integration/QueryIT.java
index 46db8cd..f408a1c 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/integration/QueryIT.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/integration/QueryIT.java
@@ -33,6 +33,7 @@ import org.openrdf.model.vocabulary.XMLSchema;
import org.openrdf.query.BindingSet;
import org.openrdf.query.impl.BindingImpl;
+import com.google.common.base.Optional;
import com.google.common.collect.Sets;
import mvm.rya.api.domain.RyaStatement;
@@ -81,7 +82,7 @@ public class QueryIT extends ITBase {
new CreatePcj().withRyaIntegration(fluoClient, RYA_TABLE_PREFIX, ryaRepo, accumuloConn, new HashSet<VariableOrder>(), sparql);
// Stream the data into Fluo.
- new InsertTriples().insert(fluoClient, streamedTriples);
+ new InsertTriples().insert(fluoClient, streamedTriples, Optional.<String>absent());
// Verify the end results of the query match the expected results.
fluo.waitForObservers();
@@ -163,7 +164,7 @@ public class QueryIT extends ITBase {
new CreatePcj().withRyaIntegration(fluoClient, RYA_TABLE_PREFIX, ryaRepo, accumuloConn, new HashSet<VariableOrder>(), sparql);
// Stream the data into Fluo.
- new InsertTriples().insert(fluoClient, streamedTriples);
+ new InsertTriples().insert(fluoClient, streamedTriples, Optional.<String>absent());
// Verify the end results of the query match the expected results.
fluo.waitForObservers();
@@ -224,7 +225,7 @@ public class QueryIT extends ITBase {
new CreatePcj().withRyaIntegration(fluoClient, RYA_TABLE_PREFIX, ryaRepo, accumuloConn, new HashSet<VariableOrder>(), sparql);
// Stream the data into Fluo.
- new InsertTriples().insert(fluoClient, streamedTriples);
+ new InsertTriples().insert(fluoClient, streamedTriples, Optional.<String>absent());
// Verify the end results of the query match the expected results.
fluo.waitForObservers();
@@ -268,7 +269,7 @@ public class QueryIT extends ITBase {
new CreatePcj().withRyaIntegration(fluoClient, RYA_TABLE_PREFIX, ryaRepo, accumuloConn, new HashSet<VariableOrder>(), sparql);
// Stream the data into Fluo.
- new InsertTriples().insert(fluoClient, streamedTriples);
+ new InsertTriples().insert(fluoClient, streamedTriples, Optional.<String>absent());
// Verify the end results of the query match the expected results.
fluo.waitForObservers();
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c53b5402/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/integration/RyaExportIT.java
----------------------------------------------------------------------
diff --git a/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/integration/RyaExportIT.java b/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/integration/RyaExportIT.java
index ee3fffd..b75e624 100644
--- a/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/integration/RyaExportIT.java
+++ b/extras/rya.pcj.fluo/pcj.fluo.integration/src/test/java/org/apache/rya/indexing/pcj/fluo/integration/RyaExportIT.java
@@ -43,6 +43,7 @@ import org.openrdf.model.impl.URIImpl;
import org.openrdf.query.BindingSet;
import org.openrdf.query.impl.BindingImpl;
+import com.google.common.base.Optional;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
@@ -50,7 +51,6 @@ import com.google.common.collect.Sets;
import io.fluo.api.client.Snapshot;
import io.fluo.api.data.Bytes;
import mvm.rya.api.domain.RyaStatement;
-import mvm.rya.api.resolver.RyaTypeResolverException;
import mvm.rya.indexing.external.tupleSet.AccumuloPcjSerializer;
import mvm.rya.indexing.external.tupleSet.BindingSetConverter.BindingSetConversionException;
import mvm.rya.indexing.external.tupleSet.PcjTables;
@@ -66,7 +66,7 @@ import mvm.rya.indexing.external.tupleSet.PcjTables.VariableOrder;
public class RyaExportIT extends ITBase {
private static final AccumuloPcjSerializer converter = new AccumuloPcjSerializer();
-
+
/**
* Configure the export observer to use the Mini Accumulo instance as the
* export destination for new PCJ results.
@@ -138,7 +138,7 @@ public class RyaExportIT extends ITBase {
new CreatePcj().withRyaIntegration(fluoClient, RYA_TABLE_PREFIX, ryaRepo, accumuloConn, new HashSet<VariableOrder>(), sparql);
// Stream the data into Fluo.
- new InsertTriples().insert(fluoClient, streamedTriples);
+ new InsertTriples().insert(fluoClient, streamedTriples, Optional.<String>absent());
// Fetch the exported results from Accumulo once the observers finish working.
fluo.waitForObservers();