RYA-303 Mongo PCJ Support. Closes #172. Added implementation of MongoPcjStorage Pcj serialization in mongo added ITs for each interactor. removed optional from RyaCommands create/delete/batch updated
Project: http://git-wip-us.apache.org/repos/asf/incubator-rya/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-rya/commit/c826ffea Tree: http://git-wip-us.apache.org/repos/asf/incubator-rya/tree/c826ffea Diff: http://git-wip-us.apache.org/repos/asf/incubator-rya/diff/c826ffea Branch: refs/heads/master Commit: c826ffea4a9046355ebae28c131042afba797918 Parents: 9f61101 Author: isper3at <[email protected]> Authored: Fri May 12 16:40:44 2017 -0700 Committer: caleb <[email protected]> Committed: Fri Jan 12 16:23:45 2018 -0500 ---------------------------------------------------------------------- .../rya/api/client/CreatePeriodicPCJ.java | 8 +- .../org/apache/rya/api/client/RyaClient.java | 19 +- .../rya/api/instance/ConfigurationFields.java | 1 + .../api/instance/RyaDetailsToConfiguration.java | 28 +- .../rya/api/resolver/RyaToRdfConversions.java | 6 +- .../rya/accumulo/RyaTestInstanceRule.java | 7 +- .../apache/rya/mongodb/MongoDBQueryEngine.java | 1 - .../mongodb/instance/MongoDetailsAdapter.java | 44 +- .../org/apache/rya/mongodb/MongoITBase.java | 2 +- .../instance/MongoDetailsAdapterTest.java | 3 - .../instance/MongoRyaDetailsRepositoryIT.java | 152 +- .../accumulo/AccumuloRyaClientFactory.java | 6 +- .../api/client/mongo/MongoBatchUpdatePCJ.java | 242 ++ .../rya/api/client/mongo/MongoCreatePCJ.java | 77 + .../rya/api/client/mongo/MongoDeletePCJ.java | 68 + .../rya/api/client/mongo/MongoInstall.java | 38 +- .../api/client/mongo/MongoRyaClientFactory.java | 6 +- .../rya/indexing/accumulo/ConfigUtils.java | 20 +- .../rya/indexing/accumulo/ConfigUtils.java.orig | 529 ----- .../external/PrecomputedJoinIndexerConfig.java | 8 +- .../mongodb/pcj/MongoPcjIndexSetProvider.java | 126 ++ .../indexing/mongodb/pcj/MongoPcjQueryNode.java | 167 ++ .../pcj/matching/AccumuloIndexSetProvider.java | 221 -- .../rya/indexing/pcj/matching/PCJOptimizer.java | 40 +- .../pcj/matching/PCJOptimizerUtilities.java | 87 +- .../provider/AbstractPcjIndexSetProvider.java | 159 ++ .../provider/AccumuloIndexSetProvider.java | 139 ++ .../accumulo/AccumuloBatchUpdatePCJIT.java | 2 +- .../api/client/accumulo/AccumuloInstallIT.java | 66 +- .../client/accumulo/AccumuloUninstallIT.java | 18 +- .../api/client/mongo/MongoBatchUpdatePCJIT.java | 148 ++ .../rya/api/client/mongo/MongoCreatePCJIT.java | 122 + .../rya/api/client/mongo/MongoDeletePCJIT.java | 93 + .../client/mongo/MongoGetInstanceDetailsIT.java | 12 +- .../rya/api/client/mongo/MongoInstallIT.java | 51 +- .../IndexPlanValidatorTest.java | 1453 ++++++------ .../ThreshholdPlanSelectorTest.java | 5 +- .../external/AccumuloConstantPcjIT.java | 4 +- .../rya/indexing/external/AccumuloPcjIT.java | 30 +- .../indexing/external/PCJOptionalTestIT.java | 27 +- .../external/PcjIntegrationTestingUtil.java | 168 +- .../external/PrecompJoinOptimizerIT.java | 16 +- .../external/PrecompJoinOptimizerTest.java | 792 +++---- .../external/PrecompJoinOptimizerTest2.java | 2121 +++++++++--------- .../PrecompJoinOptimizerVarToConstTest.java | 13 +- .../rya/indexing/mongo/MongoPCJIndexIT.java | 233 ++ .../indexing/mongo/MongoPcjIntegrationTest.java | 374 +++ .../indexing/pcj/matching/PCJOptimizerTest.java | 314 +-- .../periodic/KafkaLatencyBenchmark.java | 10 +- .../benchmark/query/PCJOptimizerBenchmark.java | 18 +- .../benchmark/query/QueryBenchmarkRunIT.java | 2 +- extras/rya.indexing.pcj/pom.xml | 10 + .../storage/mongo/MongoBindingSetConverter.java | 56 + .../pcj/storage/mongo/MongoPcjDocuments.java | 445 ++++ .../pcj/storage/mongo/MongoPcjStorage.java | 171 ++ .../storage/mongo/MongoPcjDocumentsTest.java | 91 + .../pcj/storage/mongo/MongoPcjStorageIT.java | 348 +++ .../mongo/PcjDocumentsIntegrationTest.java | 460 ++++ .../storage/mongo/PcjDocumentsWithMockTest.java | 98 + .../pcj/fluo/integration/CreateDeleteIT.java | 2 +- .../indexing/pcj/fluo/integration/QueryIT.java | 12 +- .../pcj/fluo/visibility/PcjVisibilityIT.java | 2 +- .../pcj/fluo/test/base/KafkaExportITBase.java | 2 +- .../pcj/functions/geo/GeoFunctionsIT.java | 2 +- .../org/apache/rya/shell/RyaAdminCommands.java | 49 +- .../apache/rya/shell/util/InstallPrompt.java | 25 +- .../rya/shell/util/RyaDetailsFormatter.java | 62 +- .../org/apache/rya/shell/MongoRyaShellIT.java | 13 - .../apache/rya/shell/RyaAdminCommandsTest.java | 94 +- .../rya/shell/util/RyaDetailsFormatterTest.java | 8 +- 70 files changed, 6733 insertions(+), 3513 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c826ffea/common/rya.api/src/main/java/org/apache/rya/api/client/CreatePeriodicPCJ.java ---------------------------------------------------------------------- diff --git a/common/rya.api/src/main/java/org/apache/rya/api/client/CreatePeriodicPCJ.java b/common/rya.api/src/main/java/org/apache/rya/api/client/CreatePeriodicPCJ.java index 7c006d0..a046b6f 100644 --- a/common/rya.api/src/main/java/org/apache/rya/api/client/CreatePeriodicPCJ.java +++ b/common/rya.api/src/main/java/org/apache/rya/api/client/CreatePeriodicPCJ.java @@ -19,15 +19,15 @@ package org.apache.rya.api.client; /** - * This class creates new PeriodicPCJ for a given Rya instance. + * This class creates new PeriodicPCJ for a given Rya instance. */ public interface CreatePeriodicPCJ { /** * Creates a new PeriodicPCJ for a given Rya instance. The provided periodicTopic and bootStrapServers are used for - * registering new PeriodiNotifications with the underlying notification registration service. Typically, the + * registering new Periodic Notifications with the underlying notification registration service. Typically, the * bootStrapServers are the IP for the KafkaBrokers. - * + * * @param instanceName - Rya instance to connect to * @param sparql - SPARQL query registered with the Periodic Service * @param periodicTopic - Kafka topic that new PeriodicNotifications are exported to for registration with the @@ -36,5 +36,5 @@ public interface CreatePeriodicPCJ { * @return Fluo Query Id of the registered Periodic Query */ public String createPeriodicPCJ(String instanceName, String sparql, String periodicTopic, String bootStrapServers) throws RyaClientException; - + } http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c826ffea/common/rya.api/src/main/java/org/apache/rya/api/client/RyaClient.java ---------------------------------------------------------------------- diff --git a/common/rya.api/src/main/java/org/apache/rya/api/client/RyaClient.java b/common/rya.api/src/main/java/org/apache/rya/api/client/RyaClient.java index a14e300..92b18a1 100644 --- a/common/rya.api/src/main/java/org/apache/rya/api/client/RyaClient.java +++ b/common/rya.api/src/main/java/org/apache/rya/api/client/RyaClient.java @@ -34,12 +34,12 @@ import net.jcip.annotations.Immutable; public class RyaClient { // Administrative functions. private final Install install; - private final Optional<CreatePCJ> createPcj; - private final Optional<DeletePCJ> deletePcj; + private final CreatePCJ createPcj; + private final DeletePCJ deletePcj; private final Optional<CreatePeriodicPCJ> createPeriodicPcj; private final Optional<DeletePeriodicPCJ> deletePeriodicPcj; private final Optional<ListIncrementalQueries> listIncrementalQueries; - private final Optional<BatchUpdatePCJ> bactchUpdatePCJ; + private final BatchUpdatePCJ bactchUpdatePCJ; private final GetInstanceDetails getInstanceDetails; private final InstanceExists instanceExists; private final ListInstances listInstances; @@ -55,12 +55,11 @@ public class RyaClient { */ public RyaClient( final Install install, - final Optional<CreatePCJ> createPcj, - final Optional<DeletePCJ> deletePcj, + final CreatePCJ createPcj, final DeletePCJ deletePcj, final Optional<CreatePeriodicPCJ> createPeriodicPcj, final Optional<DeletePeriodicPCJ> deletePeriodicPcj, final Optional<ListIncrementalQueries> listIncrementalQueries, - final Optional<BatchUpdatePCJ> batchUpdatePcj, + final BatchUpdatePCJ batchUpdatePcj, final GetInstanceDetails getInstanceDetails, final InstanceExists instanceExists, final ListInstances listInstances, @@ -76,7 +75,7 @@ public class RyaClient { this.createPeriodicPcj = requireNonNull(createPeriodicPcj); this.deletePeriodicPcj = requireNonNull(deletePeriodicPcj); this.listIncrementalQueries = requireNonNull(listIncrementalQueries); - this.bactchUpdatePCJ = requireNonNull(batchUpdatePcj); + bactchUpdatePCJ = requireNonNull(batchUpdatePcj); this.getInstanceDetails = requireNonNull(getInstanceDetails); this.instanceExists = requireNonNull(instanceExists); this.listInstances = requireNonNull(listInstances); @@ -99,7 +98,7 @@ public class RyaClient { * @return An instance of {@link CreatePCJ} that is connected to a Rya storage * if the Rya instance supports PCJ indexing. */ - public Optional<CreatePCJ> getCreatePCJ() { + public CreatePCJ getCreatePCJ() { return createPcj; } @@ -107,7 +106,7 @@ public class RyaClient { * @return An instance of {@link DeletePCJ} that is connected to a Rya storage * if the Rya instance supports PCJ indexing. */ - public Optional<DeletePCJ> getDeletePCJ() { + public DeletePCJ getDeletePCJ() { return deletePcj; } @@ -137,7 +136,7 @@ public class RyaClient { * @return An instance of {@link BatchUpdatePCJ} that is connect to a Rya storage * if the Rya instance supports PCJ indexing. */ - public Optional<BatchUpdatePCJ> getBatchUpdatePCJ() { + public BatchUpdatePCJ getBatchUpdatePCJ() { return bactchUpdatePCJ; } http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c826ffea/common/rya.api/src/main/java/org/apache/rya/api/instance/ConfigurationFields.java ---------------------------------------------------------------------- diff --git a/common/rya.api/src/main/java/org/apache/rya/api/instance/ConfigurationFields.java b/common/rya.api/src/main/java/org/apache/rya/api/instance/ConfigurationFields.java index 086c3fb..4173dd3 100644 --- a/common/rya.api/src/main/java/org/apache/rya/api/instance/ConfigurationFields.java +++ b/common/rya.api/src/main/java/org/apache/rya/api/instance/ConfigurationFields.java @@ -32,6 +32,7 @@ class ConfigurationFields { static final String USE_TEMPORAL = "sc.use_temporal"; static final String USE_ENTITY = "sc.use_entity"; static final String USE_PCJ_UPDATER = "sc.use.updater"; + static final String USE_MONGO = "sc.useMongo"; static final String FLUO_APP_NAME = "rya.indexing.pcj.fluo.fluoAppName"; static final String PCJ_UPDATER_TYPE = "rya.indexing.pcj.updaterType"; static final String PCJ_STORAGE_TYPE = "rya.indexing.pcj.storageType"; http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c826ffea/common/rya.api/src/main/java/org/apache/rya/api/instance/RyaDetailsToConfiguration.java ---------------------------------------------------------------------- diff --git a/common/rya.api/src/main/java/org/apache/rya/api/instance/RyaDetailsToConfiguration.java b/common/rya.api/src/main/java/org/apache/rya/api/instance/RyaDetailsToConfiguration.java index 2605906..a8d47ba 100644 --- a/common/rya.api/src/main/java/org/apache/rya/api/instance/RyaDetailsToConfiguration.java +++ b/common/rya.api/src/main/java/org/apache/rya/api/instance/RyaDetailsToConfiguration.java @@ -51,17 +51,29 @@ public class RyaDetailsToConfiguration { checkAndSet(conf, ConfigurationFields.USE_ENTITY, details.getEntityCentricIndexDetails().isEnabled()); checkAndSet(conf, ConfigurationFields.USE_FREETEXT, details.getFreeTextIndexDetails().isEnabled()); - //RYA-215 checkAndSet(conf, ConfigurationFields.USE_GEO, details.getGeoIndexDetails().isEnabled()); + //RYA-215 checkAndSet(conf, ConfigurationFields.USE_GEO, details.getGeoIndexDetails().isEnabled()); checkAndSet(conf, ConfigurationFields.USE_TEMPORAL, details.getTemporalIndexDetails().isEnabled()); final PCJIndexDetails pcjDetails = details.getPCJIndexDetails(); - if (pcjDetails.isEnabled() && pcjDetails.getFluoDetails().isPresent()) { - checkAndSet(conf, ConfigurationFields.USE_PCJ_UPDATER, true); - conf.set(ConfigurationFields.FLUO_APP_NAME, pcjDetails.getFluoDetails().get().getUpdateAppName()); - conf.set(ConfigurationFields.PCJ_UPDATER_TYPE, "FLUO"); - conf.set(ConfigurationFields.PCJ_STORAGE_TYPE, "ACCUMULO"); + + if(conf.getBoolean(ConfigurationFields.USE_MONGO, false) ) { + if(pcjDetails.isEnabled()) { + conf.set(ConfigurationFields.PCJ_STORAGE_TYPE, "MONGO"); + //mongo does not currently support pcj updaters + checkAndSet(conf, ConfigurationFields.USE_PCJ_UPDATER, false); + conf.set(ConfigurationFields.PCJ_UPDATER_TYPE, "NO_UPDATE"); + } } else { - checkAndSet(conf, ConfigurationFields.USE_PCJ_UPDATER, false); - conf.set(ConfigurationFields.PCJ_UPDATER_TYPE, "NO_UPDATE"); + if (pcjDetails.isEnabled() ) { + conf.set(ConfigurationFields.PCJ_STORAGE_TYPE, "ACCUMULO"); + if (pcjDetails.getFluoDetails().isPresent()) { + checkAndSet(conf, ConfigurationFields.USE_PCJ_UPDATER, true); + conf.set(ConfigurationFields.FLUO_APP_NAME, pcjDetails.getFluoDetails().get().getUpdateAppName()); + conf.set(ConfigurationFields.PCJ_UPDATER_TYPE, "FLUO"); + } else { + checkAndSet(conf, ConfigurationFields.USE_PCJ_UPDATER, false); + conf.set(ConfigurationFields.PCJ_UPDATER_TYPE, "NO_UPDATE"); + } + } } } http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c826ffea/common/rya.api/src/main/java/org/apache/rya/api/resolver/RyaToRdfConversions.java ---------------------------------------------------------------------- diff --git a/common/rya.api/src/main/java/org/apache/rya/api/resolver/RyaToRdfConversions.java b/common/rya.api/src/main/java/org/apache/rya/api/resolver/RyaToRdfConversions.java index a59ba23..51062c9 100644 --- a/common/rya.api/src/main/java/org/apache/rya/api/resolver/RyaToRdfConversions.java +++ b/common/rya.api/src/main/java/org/apache/rya/api/resolver/RyaToRdfConversions.java @@ -43,6 +43,10 @@ public class RyaToRdfConversions { public static URI convertURI(RyaURI uri) { return new URIImpl(uri.getData()); } + + private static URI convertURI(RyaType value) { + return new URIImpl(value.getData()); + } public static Literal convertLiteral(RyaType literal) { if (XMLSchema.STRING.equals(literal.getDataType())) { @@ -55,7 +59,7 @@ public class RyaToRdfConversions { public static Value convertValue(RyaType value) { //assuming either uri or Literal here - return (value instanceof RyaURI) ? convertURI((RyaURI) value) : convertLiteral(value); + return (value instanceof RyaURI || value.getDataType().equals(XMLSchema.ANYURI)) ? convertURI(value) : convertLiteral(value); } public static Statement convertStatement(RyaStatement statement) { http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c826ffea/dao/accumulo.rya/src/test/java/org/apache/rya/accumulo/RyaTestInstanceRule.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/test/java/org/apache/rya/accumulo/RyaTestInstanceRule.java b/dao/accumulo.rya/src/test/java/org/apache/rya/accumulo/RyaTestInstanceRule.java index 38b4b43..3bbceb9 100644 --- a/dao/accumulo.rya/src/test/java/org/apache/rya/accumulo/RyaTestInstanceRule.java +++ b/dao/accumulo.rya/src/test/java/org/apache/rya/accumulo/RyaTestInstanceRule.java @@ -1,9 +1,8 @@ package org.apache.rya.accumulo; -import com.google.common.base.Optional; +import java.util.concurrent.atomic.AtomicInteger; + import org.apache.rya.accumulo.instance.AccumuloRyaInstanceDetailsRepository; -import org.apache.rya.api.client.Install.InstallConfiguration; -import org.apache.rya.api.client.RyaClient; import org.apache.rya.api.instance.RyaDetails; import org.apache.rya.api.instance.RyaDetails.EntityCentricIndexDetails; import org.apache.rya.api.instance.RyaDetails.FreeTextIndexDetails; @@ -14,7 +13,7 @@ import org.apache.rya.api.instance.RyaDetails.TemporalIndexDetails; import org.apache.rya.api.instance.RyaDetailsRepository; import org.junit.rules.ExternalResource; -import java.util.concurrent.atomic.AtomicInteger; +import com.google.common.base.Optional; /** * Licensed to the Apache Software Foundation (ASF) under one http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c826ffea/dao/mongodb.rya/src/main/java/org/apache/rya/mongodb/MongoDBQueryEngine.java ---------------------------------------------------------------------- diff --git a/dao/mongodb.rya/src/main/java/org/apache/rya/mongodb/MongoDBQueryEngine.java b/dao/mongodb.rya/src/main/java/org/apache/rya/mongodb/MongoDBQueryEngine.java index e01fa34..9ddb15a 100644 --- a/dao/mongodb.rya/src/main/java/org/apache/rya/mongodb/MongoDBQueryEngine.java +++ b/dao/mongodb.rya/src/main/java/org/apache/rya/mongodb/MongoDBQueryEngine.java @@ -154,6 +154,5 @@ public class MongoDBQueryEngine implements RyaQueryEngine<StatefulMongoDBRdfConf @Override public void close() throws IOException { -// if (mongoClient != null){ mongoClient.close(); } } } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c826ffea/dao/mongodb.rya/src/main/java/org/apache/rya/mongodb/instance/MongoDetailsAdapter.java ---------------------------------------------------------------------- diff --git a/dao/mongodb.rya/src/main/java/org/apache/rya/mongodb/instance/MongoDetailsAdapter.java b/dao/mongodb.rya/src/main/java/org/apache/rya/mongodb/instance/MongoDetailsAdapter.java index 3b77a88..8010808 100644 --- a/dao/mongodb.rya/src/main/java/org/apache/rya/mongodb/instance/MongoDetailsAdapter.java +++ b/dao/mongodb.rya/src/main/java/org/apache/rya/mongodb/instance/MongoDetailsAdapter.java @@ -29,7 +29,6 @@ import org.apache.rya.api.instance.RyaDetails.EntityCentricIndexDetails; import org.apache.rya.api.instance.RyaDetails.FreeTextIndexDetails; import org.apache.rya.api.instance.RyaDetails.JoinSelectivityDetails; import org.apache.rya.api.instance.RyaDetails.PCJIndexDetails; -import org.apache.rya.api.instance.RyaDetails.PCJIndexDetails.FluoDetails; import org.apache.rya.api.instance.RyaDetails.PCJIndexDetails.PCJDetails; import org.apache.rya.api.instance.RyaDetails.PCJIndexDetails.PCJDetails.PCJUpdateStrategy; import org.apache.rya.api.instance.RyaDetails.ProspectorDetails; @@ -100,13 +99,13 @@ public class MongoDetailsAdapter { public static BasicDBObject toDBObject(final RyaDetails details) { Preconditions.checkNotNull(details); final BasicDBObjectBuilder builder = BasicDBObjectBuilder.start() - .add(INSTANCE_KEY, details.getRyaInstanceName()) - .add(VERSION_KEY, details.getRyaVersion()) - .add(ENTITY_DETAILS_KEY, details.getEntityCentricIndexDetails().isEnabled()) - //RYA-215 .add(GEO_DETAILS_KEY, details.getGeoIndexDetails().isEnabled()) - .add(PCJ_DETAILS_KEY, toDBObject(details.getPCJIndexDetails())) - .add(TEMPORAL_DETAILS_KEY, details.getTemporalIndexDetails().isEnabled()) - .add(FREETEXT_DETAILS_KEY, details.getFreeTextIndexDetails().isEnabled()); + .add(INSTANCE_KEY, details.getRyaInstanceName()) + .add(VERSION_KEY, details.getRyaVersion()) + .add(ENTITY_DETAILS_KEY, details.getEntityCentricIndexDetails().isEnabled()) + //RYA-215 .add(GEO_DETAILS_KEY, details.getGeoIndexDetails().isEnabled()) + .add(PCJ_DETAILS_KEY, toDBObject(details.getPCJIndexDetails())) + .add(TEMPORAL_DETAILS_KEY, details.getTemporalIndexDetails().isEnabled()) + .add(FREETEXT_DETAILS_KEY, details.getFreeTextIndexDetails().isEnabled()); if(details.getProspectorDetails().getLastUpdated().isPresent()) { builder.add(PROSPECTOR_DETAILS_KEY, details.getProspectorDetails().getLastUpdated().get()); } @@ -124,11 +123,6 @@ public class MongoDetailsAdapter { // Is Enabled builder.add(PCJ_ENABLED_KEY, pcjIndexDetails.isEnabled()); - // Fluo Details if present. - if(pcjIndexDetails.getFluoDetails().isPresent()) { - builder.add(PCJ_FLUO_KEY, pcjIndexDetails.getFluoDetails().get().getUpdateAppName()); - } - // Add the PCJDetail objects. final List<DBObject> pcjDetailsList = new ArrayList<>(); for(final PCJDetails pcjDetails : pcjIndexDetails.getPCJDetails().values()) { @@ -163,17 +157,17 @@ public class MongoDetailsAdapter { public static RyaDetails toRyaDetails(final DBObject mongoObj) throws MalformedRyaDetailsException { final BasicDBObject basicObj = (BasicDBObject) mongoObj; try { - return RyaDetails.builder() - .setRyaInstanceName(basicObj.getString(INSTANCE_KEY)) - .setRyaVersion(basicObj.getString(VERSION_KEY)) - .setEntityCentricIndexDetails(new EntityCentricIndexDetails(basicObj.getBoolean(ENTITY_DETAILS_KEY))) - //RYA-215 .setGeoIndexDetails(new GeoIndexDetails(basicObj.getBoolean(GEO_DETAILS_KEY))) - .setPCJIndexDetails(getPCJIndexDetails(basicObj)) - .setTemporalIndexDetails(new TemporalIndexDetails(basicObj.getBoolean(TEMPORAL_DETAILS_KEY))) - .setFreeTextDetails(new FreeTextIndexDetails(basicObj.getBoolean(FREETEXT_DETAILS_KEY))) - .setProspectorDetails(new ProspectorDetails(Optional.<Date>fromNullable(basicObj.getDate(PROSPECTOR_DETAILS_KEY)))) - .setJoinSelectivityDetails(new JoinSelectivityDetails(Optional.<Date>fromNullable(basicObj.getDate(JOIN_SELECTIVITY_DETAILS_KEY)))) - .build(); + return RyaDetails.builder() + .setRyaInstanceName(basicObj.getString(INSTANCE_KEY)) + .setRyaVersion(basicObj.getString(VERSION_KEY)) + .setEntityCentricIndexDetails(new EntityCentricIndexDetails(basicObj.getBoolean(ENTITY_DETAILS_KEY))) + //RYA-215 .setGeoIndexDetails(new GeoIndexDetails(basicObj.getBoolean(GEO_DETAILS_KEY))) + .setPCJIndexDetails(getPCJIndexDetails(basicObj)) + .setTemporalIndexDetails(new TemporalIndexDetails(basicObj.getBoolean(TEMPORAL_DETAILS_KEY))) + .setFreeTextDetails(new FreeTextIndexDetails(basicObj.getBoolean(FREETEXT_DETAILS_KEY))) + .setProspectorDetails(new ProspectorDetails(Optional.<Date>fromNullable(basicObj.getDate(PROSPECTOR_DETAILS_KEY)))) + .setJoinSelectivityDetails(new JoinSelectivityDetails(Optional.<Date>fromNullable(basicObj.getDate(JOIN_SELECTIVITY_DETAILS_KEY)))) + .build(); } catch(final Exception e) { throw new MalformedRyaDetailsException("Failed to make RyaDetail from Mongo Object, it is malformed.", e); } @@ -186,7 +180,7 @@ public class MongoDetailsAdapter { if (!pcjIndexDBO.getBoolean(PCJ_ENABLED_KEY)) { pcjBuilder.setEnabled(false); } else { - pcjBuilder.setEnabled(true).setFluoDetails(new FluoDetails(pcjIndexDBO.getString(PCJ_FLUO_KEY))); + pcjBuilder.setEnabled(true);//no fluo details to set since mongo has no fluo support final BasicDBList pcjs = (BasicDBList) pcjIndexDBO.get(PCJ_PCJS_KEY); if (pcjs != null) { for (int ii = 0; ii < pcjs.size(); ii++) { http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c826ffea/dao/mongodb.rya/src/test/java/org/apache/rya/mongodb/MongoITBase.java ---------------------------------------------------------------------- diff --git a/dao/mongodb.rya/src/test/java/org/apache/rya/mongodb/MongoITBase.java b/dao/mongodb.rya/src/test/java/org/apache/rya/mongodb/MongoITBase.java index 1aaf6c2..bd3870f 100644 --- a/dao/mongodb.rya/src/test/java/org/apache/rya/mongodb/MongoITBase.java +++ b/dao/mongodb.rya/src/test/java/org/apache/rya/mongodb/MongoITBase.java @@ -43,7 +43,7 @@ public class MongoITBase { final MongoDBRdfConfiguration conf = new MongoDBRdfConfiguration( new Configuration() ); conf.setBoolean("sc.useMongo", true); conf.setTablePrefix("test_"); - conf.setMongoDBName("testDB"); + conf.setMongoDBName(conf.getRyaInstanceName()); conf.setMongoHostname(EmbeddedMongoSingleton.getMongodConfig().net().getServerAddress().getHostAddress()); conf.setMongoPort(Integer.toString(EmbeddedMongoSingleton.getMongodConfig().net().getPort())); http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c826ffea/dao/mongodb.rya/src/test/java/org/apache/rya/mongodb/instance/MongoDetailsAdapterTest.java ---------------------------------------------------------------------- diff --git a/dao/mongodb.rya/src/test/java/org/apache/rya/mongodb/instance/MongoDetailsAdapterTest.java b/dao/mongodb.rya/src/test/java/org/apache/rya/mongodb/instance/MongoDetailsAdapterTest.java index 875f4e5..0ea9456 100644 --- a/dao/mongodb.rya/src/test/java/org/apache/rya/mongodb/instance/MongoDetailsAdapterTest.java +++ b/dao/mongodb.rya/src/test/java/org/apache/rya/mongodb/instance/MongoDetailsAdapterTest.java @@ -85,7 +85,6 @@ public class MongoDetailsAdapterTest { //RYA-215 + "geoDetails : true," + "pcjDetails : {" + "enabled : true ," - + "fluoName : \"fluo\"," + "pcjs : [ " + "{" + "id : \"pcj_0\"," @@ -150,7 +149,6 @@ public class MongoDetailsAdapterTest { .setPCJIndexDetails( PCJIndexDetails.builder() .setEnabled(true) - .setFluoDetails(new FluoDetails("fluo")) .addPCJDetails( PCJDetails.builder() .setId("pcj_0") @@ -242,7 +240,6 @@ public class MongoDetailsAdapterTest { //RYA-215 + "geoDetails : false," + "pcjDetails : {" + "enabled : true," - + "fluoName : \"fluo\"," + "pcjs : [ ]" + "}," + "temporalDetails : false," http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c826ffea/dao/mongodb.rya/src/test/java/org/apache/rya/mongodb/instance/MongoRyaDetailsRepositoryIT.java ---------------------------------------------------------------------- diff --git a/dao/mongodb.rya/src/test/java/org/apache/rya/mongodb/instance/MongoRyaDetailsRepositoryIT.java b/dao/mongodb.rya/src/test/java/org/apache/rya/mongodb/instance/MongoRyaDetailsRepositoryIT.java index dd5088e..5321db5 100644 --- a/dao/mongodb.rya/src/test/java/org/apache/rya/mongodb/instance/MongoRyaDetailsRepositoryIT.java +++ b/dao/mongodb.rya/src/test/java/org/apache/rya/mongodb/instance/MongoRyaDetailsRepositoryIT.java @@ -64,27 +64,26 @@ public class MongoRyaDetailsRepositoryIT extends MongoITBase { // Create the metadata object the repository will be initialized with. final RyaDetails details = RyaDetails.builder() - .setRyaInstanceName(instanceName) - .setRyaVersion("1.2.3.4") - .setEntityCentricIndexDetails( new EntityCentricIndexDetails(true) ) - //RYA-215 .setGeoIndexDetails( new GeoIndexDetails(true) ) - .setTemporalIndexDetails( new TemporalIndexDetails(true) ) - .setFreeTextDetails( new FreeTextIndexDetails(true) ) - .setPCJIndexDetails( - PCJIndexDetails.builder() + .setRyaInstanceName(instanceName) + .setRyaVersion("1.2.3.4") + .setEntityCentricIndexDetails( new EntityCentricIndexDetails(true) ) + //RYA-215 .setGeoIndexDetails( new GeoIndexDetails(true) ) + .setTemporalIndexDetails( new TemporalIndexDetails(true) ) + .setFreeTextDetails( new FreeTextIndexDetails(true) ) + .setPCJIndexDetails( + PCJIndexDetails.builder() .setEnabled(true) - .setFluoDetails( new FluoDetails("test_instance_rya_pcj_updater") ) .addPCJDetails( PCJDetails.builder() - .setId("pcj 1") - .setUpdateStrategy(PCJUpdateStrategy.BATCH) - .setLastUpdateTime( new Date() )) + .setId("pcj 1") + .setUpdateStrategy(PCJUpdateStrategy.BATCH) + .setLastUpdateTime( new Date() )) .addPCJDetails( PCJDetails.builder() - .setId("pcj 2"))) - .setProspectorDetails( new ProspectorDetails(Optional.of(new Date())) ) - .setJoinSelectivityDetails( new JoinSelectivityDetails(Optional.of(new Date())) ) - .build(); + .setId("pcj 2"))) + .setProspectorDetails( new ProspectorDetails(Optional.of(new Date())) ) + .setJoinSelectivityDetails( new JoinSelectivityDetails(Optional.of(new Date())) ) + .build(); // Setup the repository that will be tested using a mock instance of MongoDB. final RyaDetailsRepository repo = new MongoRyaInstanceDetailsRepository(client, instanceName); @@ -105,27 +104,27 @@ public class MongoRyaDetailsRepositoryIT extends MongoITBase { // Create the metadata object the repository will be initialized with. final RyaDetails details = RyaDetails.builder() - .setRyaInstanceName(instanceName) - .setRyaVersion("1.2.3.4") - .setEntityCentricIndexDetails( new EntityCentricIndexDetails(true) ) - //RYA-215 .setGeoIndexDetails( new GeoIndexDetails(true) ) - .setTemporalIndexDetails( new TemporalIndexDetails(true) ) - .setFreeTextDetails( new FreeTextIndexDetails(true) ) - .setPCJIndexDetails( - PCJIndexDetails.builder() + .setRyaInstanceName(instanceName) + .setRyaVersion("1.2.3.4") + .setEntityCentricIndexDetails( new EntityCentricIndexDetails(true) ) + //RYA-215 .setGeoIndexDetails( new GeoIndexDetails(true) ) + .setTemporalIndexDetails( new TemporalIndexDetails(true) ) + .setFreeTextDetails( new FreeTextIndexDetails(true) ) + .setPCJIndexDetails( + PCJIndexDetails.builder() .setEnabled(true) .setFluoDetails( new FluoDetails("test_instance_rya_pcj_updater") ) .addPCJDetails( PCJDetails.builder() - .setId("pcj 1") - .setUpdateStrategy(PCJUpdateStrategy.BATCH) - .setLastUpdateTime( new Date() )) + .setId("pcj 1") + .setUpdateStrategy(PCJUpdateStrategy.BATCH) + .setLastUpdateTime( new Date() )) .addPCJDetails( PCJDetails.builder() - .setId("pcj 2"))) - .setProspectorDetails( new ProspectorDetails(Optional.of(new Date())) ) - .setJoinSelectivityDetails( new JoinSelectivityDetails(Optional.of(new Date())) ) - .build(); + .setId("pcj 2"))) + .setProspectorDetails( new ProspectorDetails(Optional.of(new Date())) ) + .setJoinSelectivityDetails( new JoinSelectivityDetails(Optional.of(new Date())) ) + .build(); // Setup the repository that will be tested using a mock instance of MongoDB. final RyaDetailsRepository repo = new MongoRyaInstanceDetailsRepository(client, instanceName); @@ -152,27 +151,27 @@ public class MongoRyaDetailsRepositoryIT extends MongoITBase { // Create the metadata object the repository will be initialized with. final RyaDetails details = RyaDetails.builder() - .setRyaInstanceName(instanceName) - .setRyaVersion("1.2.3.4") - .setEntityCentricIndexDetails( new EntityCentricIndexDetails(true) ) - //RYA-215 .setGeoIndexDetails( new GeoIndexDetails(true) ) - .setTemporalIndexDetails( new TemporalIndexDetails(true) ) - .setFreeTextDetails( new FreeTextIndexDetails(true) ) - .setPCJIndexDetails( - PCJIndexDetails.builder() + .setRyaInstanceName(instanceName) + .setRyaVersion("1.2.3.4") + .setEntityCentricIndexDetails( new EntityCentricIndexDetails(true) ) + //RYA-215 .setGeoIndexDetails( new GeoIndexDetails(true) ) + .setTemporalIndexDetails( new TemporalIndexDetails(true) ) + .setFreeTextDetails( new FreeTextIndexDetails(true) ) + .setPCJIndexDetails( + PCJIndexDetails.builder() .setEnabled(true) .setFluoDetails( new FluoDetails("test_instance_rya_pcj_updater") ) .addPCJDetails( PCJDetails.builder() - .setId("pcj 1") - .setUpdateStrategy(PCJUpdateStrategy.BATCH) - .setLastUpdateTime( new Date() )) + .setId("pcj 1") + .setUpdateStrategy(PCJUpdateStrategy.BATCH) + .setLastUpdateTime( new Date() )) .addPCJDetails( PCJDetails.builder() - .setId("pcj 2"))) - .setProspectorDetails( new ProspectorDetails(Optional.of(new Date())) ) - .setJoinSelectivityDetails( new JoinSelectivityDetails(Optional.of(new Date())) ) - .build(); + .setId("pcj 2"))) + .setProspectorDetails( new ProspectorDetails(Optional.of(new Date())) ) + .setJoinSelectivityDetails( new JoinSelectivityDetails(Optional.of(new Date())) ) + .build(); // Setup the repository that will be tested using a mock instance of MongoDB. final RyaDetailsRepository repo = new MongoRyaInstanceDetailsRepository(client, "testInstance"); @@ -199,27 +198,26 @@ public class MongoRyaDetailsRepositoryIT extends MongoITBase { // Create the metadata object the repository will be initialized with. final RyaDetails details = RyaDetails.builder() - .setRyaInstanceName(instanceName) - .setRyaVersion("1.2.3.4") - .setEntityCentricIndexDetails( new EntityCentricIndexDetails(true) ) - //RYA-215 .setGeoIndexDetails( new GeoIndexDetails(true) ) - .setTemporalIndexDetails( new TemporalIndexDetails(true) ) - .setFreeTextDetails( new FreeTextIndexDetails(true) ) - .setPCJIndexDetails( - PCJIndexDetails.builder() + .setRyaInstanceName(instanceName) + .setRyaVersion("1.2.3.4") + .setEntityCentricIndexDetails( new EntityCentricIndexDetails(true) ) + //RYA-215 .setGeoIndexDetails( new GeoIndexDetails(true) ) + .setTemporalIndexDetails( new TemporalIndexDetails(true) ) + .setFreeTextDetails( new FreeTextIndexDetails(true) ) + .setPCJIndexDetails( + PCJIndexDetails.builder() .setEnabled(true) - .setFluoDetails( new FluoDetails("test_instance_rya_pcj_updater") ) .addPCJDetails( PCJDetails.builder() - .setId("pcj 1") - .setUpdateStrategy(PCJUpdateStrategy.BATCH) - .setLastUpdateTime( new Date() )) + .setId("pcj 1") + .setUpdateStrategy(PCJUpdateStrategy.BATCH) + .setLastUpdateTime( new Date() )) .addPCJDetails( PCJDetails.builder() - .setId("pcj 2"))) - .setProspectorDetails( new ProspectorDetails(Optional.of(new Date())) ) - .setJoinSelectivityDetails( new JoinSelectivityDetails(Optional.of(new Date())) ) - .build(); + .setId("pcj 2"))) + .setProspectorDetails( new ProspectorDetails(Optional.of(new Date())) ) + .setJoinSelectivityDetails( new JoinSelectivityDetails(Optional.of(new Date())) ) + .build(); // Setup the repository that will be tested using a mock instance of MongoDB. final RyaDetailsRepository repo = new MongoRyaInstanceDetailsRepository(client, "testInstance"); @@ -246,27 +244,27 @@ public class MongoRyaDetailsRepositoryIT extends MongoITBase { // Create the metadata object the repository will be initialized with. final RyaDetails details = RyaDetails.builder() - .setRyaInstanceName(instanceName) - .setRyaVersion("1.2.3.4") - .setEntityCentricIndexDetails( new EntityCentricIndexDetails(true) ) - //RYA-215 .setGeoIndexDetails( new GeoIndexDetails(true) ) - .setTemporalIndexDetails( new TemporalIndexDetails(true) ) - .setFreeTextDetails( new FreeTextIndexDetails(true) ) - .setPCJIndexDetails( - PCJIndexDetails.builder() + .setRyaInstanceName(instanceName) + .setRyaVersion("1.2.3.4") + .setEntityCentricIndexDetails( new EntityCentricIndexDetails(true) ) + //RYA-215 .setGeoIndexDetails( new GeoIndexDetails(true) ) + .setTemporalIndexDetails( new TemporalIndexDetails(true) ) + .setFreeTextDetails( new FreeTextIndexDetails(true) ) + .setPCJIndexDetails( + PCJIndexDetails.builder() .setEnabled(true) .setFluoDetails( new FluoDetails("test_instance_rya_pcj_updater") ) .addPCJDetails( PCJDetails.builder() - .setId("pcj 1") - .setUpdateStrategy(PCJUpdateStrategy.BATCH) - .setLastUpdateTime( new Date() )) + .setId("pcj 1") + .setUpdateStrategy(PCJUpdateStrategy.BATCH) + .setLastUpdateTime( new Date() )) .addPCJDetails( PCJDetails.builder() - .setId("pcj 2"))) - .setProspectorDetails( new ProspectorDetails(Optional.of(new Date())) ) - .setJoinSelectivityDetails( new JoinSelectivityDetails(Optional.of(new Date())) ) - .build(); + .setId("pcj 2"))) + .setProspectorDetails( new ProspectorDetails(Optional.of(new Date())) ) + .setJoinSelectivityDetails( new JoinSelectivityDetails(Optional.of(new Date())) ) + .build(); // Setup the repository that will be tested using a mock instance of MongoDB. final RyaDetailsRepository repo = new MongoRyaInstanceDetailsRepository(client, "testInstance"); http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c826ffea/extras/indexing/src/main/java/org/apache/rya/api/client/accumulo/AccumuloRyaClientFactory.java ---------------------------------------------------------------------- diff --git a/extras/indexing/src/main/java/org/apache/rya/api/client/accumulo/AccumuloRyaClientFactory.java b/extras/indexing/src/main/java/org/apache/rya/api/client/accumulo/AccumuloRyaClientFactory.java index f9df491..17fddaa 100644 --- a/extras/indexing/src/main/java/org/apache/rya/api/client/accumulo/AccumuloRyaClientFactory.java +++ b/extras/indexing/src/main/java/org/apache/rya/api/client/accumulo/AccumuloRyaClientFactory.java @@ -52,12 +52,12 @@ public class AccumuloRyaClientFactory { // Build the RyaCommands option with the initialized commands. return new RyaClient( new AccumuloInstall(connectionDetails, connector), - Optional.of(new AccumuloCreatePCJ(connectionDetails, connector)), - Optional.of(new AccumuloDeletePCJ(connectionDetails, connector)), + new AccumuloCreatePCJ(connectionDetails, connector), + new AccumuloDeletePCJ(connectionDetails, connector), Optional.of(new AccumuloCreatePeriodicPCJ(connectionDetails, connector)), Optional.of(new AccumuloDeletePeriodicPCJ(connectionDetails, connector)), Optional.of(new AccumuloListIncrementalQueries(connectionDetails, connector)), - Optional.of(new AccumuloBatchUpdatePCJ(connectionDetails, connector)), + new AccumuloBatchUpdatePCJ(connectionDetails, connector), new AccumuloGetInstanceDetails(connectionDetails, connector), new AccumuloInstanceExists(connectionDetails, connector), new AccumuloListInstances(connectionDetails, connector), http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c826ffea/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoBatchUpdatePCJ.java ---------------------------------------------------------------------- diff --git a/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoBatchUpdatePCJ.java b/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoBatchUpdatePCJ.java new file mode 100644 index 0000000..3fe99ef --- /dev/null +++ b/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoBatchUpdatePCJ.java @@ -0,0 +1,242 @@ +/** + * 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.api.client.mongo; + +import static com.google.common.base.Preconditions.checkState; +import static java.util.Objects.requireNonNull; + +import java.util.ArrayList; +import java.util.Date; +import java.util.List; + +import org.apache.accumulo.core.client.AccumuloException; +import org.apache.accumulo.core.client.AccumuloSecurityException; +import org.apache.rya.api.client.BatchUpdatePCJ; +import org.apache.rya.api.client.InstanceDoesNotExistException; +import org.apache.rya.api.client.InstanceExists; +import org.apache.rya.api.client.PCJDoesNotExistException; +import org.apache.rya.api.client.RyaClientException; +import org.apache.rya.api.instance.RyaDetails; +import org.apache.rya.api.instance.RyaDetails.PCJIndexDetails.PCJDetails; +import org.apache.rya.api.instance.RyaDetails.PCJIndexDetails.PCJDetails.PCJUpdateStrategy; +import org.apache.rya.api.instance.RyaDetailsRepository; +import org.apache.rya.api.instance.RyaDetailsRepository.NotInitializedException; +import org.apache.rya.api.instance.RyaDetailsRepository.RyaDetailsRepositoryException; +import org.apache.rya.api.instance.RyaDetailsUpdater; +import org.apache.rya.api.instance.RyaDetailsUpdater.RyaDetailsMutator; +import org.apache.rya.api.instance.RyaDetailsUpdater.RyaDetailsMutator.CouldNotApplyMutationException; +import org.apache.rya.api.model.VisibilityBindingSet; +import org.apache.rya.api.persist.RyaDAOException; +import org.apache.rya.indexing.pcj.storage.PcjMetadata; +import org.apache.rya.indexing.pcj.storage.PrecomputedJoinStorage; +import org.apache.rya.indexing.pcj.storage.PrecomputedJoinStorage.PCJStorageException; +import org.apache.rya.indexing.pcj.storage.mongo.MongoPcjStorage; +import org.apache.rya.mongodb.MongoDBRdfConfiguration; +import org.apache.rya.mongodb.instance.MongoRyaInstanceDetailsRepository; +import org.apache.rya.rdftriplestore.inference.InferenceEngineException; +import org.apache.rya.sail.config.RyaSailFactory; +import org.openrdf.query.BindingSet; +import org.openrdf.query.MalformedQueryException; +import org.openrdf.query.QueryEvaluationException; +import org.openrdf.query.QueryLanguage; +import org.openrdf.query.TupleQuery; +import org.openrdf.query.TupleQueryResultHandlerBase; +import org.openrdf.query.TupleQueryResultHandlerException; +import org.openrdf.repository.RepositoryException; +import org.openrdf.repository.sail.SailRepository; +import org.openrdf.repository.sail.SailRepositoryConnection; +import org.openrdf.sail.Sail; +import org.openrdf.sail.SailConnection; +import org.openrdf.sail.SailException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.mongodb.MongoClient; + +/** + * A Mongo implementation of {@link MongoBatchUpdatePCJ}. + * <p> + * <b>Note:</b> + * <p> + * Using this batch updater can create data leaks since the Sail layer does not + * support {@link VisibilityBindingSet}s. + */ +public class MongoBatchUpdatePCJ implements BatchUpdatePCJ { + private static final Logger log = LoggerFactory.getLogger(MongoBatchUpdatePCJ.class); + + private final MongoConnectionDetails connectionDetails; + private final MongoClient mongoClient; + private final InstanceExists instanceExists; + + /** + * Constructs an instance of {@link MongoBatchUpdatePCJ}. + * + * @param connectionDetails - Details to connect to the server. (not null) + * @param instanceExists - The interactor used to check if a Rya instance exists. (not null) + * @param mongoClient - The {@link MongoClient} to use when batch updating. (not null) + */ + public MongoBatchUpdatePCJ( + final MongoConnectionDetails connectionDetails, + final MongoClient mongoClient, + final MongoInstanceExists instanceExists) { + this.connectionDetails = requireNonNull(connectionDetails); + this.mongoClient = requireNonNull(mongoClient); + this.instanceExists = requireNonNull(instanceExists); + } + + @Override + public void batchUpdate(final String ryaInstanceName, final String pcjId) throws InstanceDoesNotExistException, PCJDoesNotExistException, RyaClientException { + requireNonNull(ryaInstanceName); + requireNonNull(pcjId); + + checkState(instanceExists.exists(ryaInstanceName), "The instance: " + ryaInstanceName + " does not exist."); + + verifyPCJState(ryaInstanceName, pcjId, mongoClient); + updatePCJResults(ryaInstanceName, pcjId, mongoClient); + updatePCJMetadata(ryaInstanceName, pcjId, mongoClient); + } + + private void verifyPCJState(final String ryaInstanceName, final String pcjId, final MongoClient client) throws RyaClientException { + try { + // Fetch the Rya instance's details. + final RyaDetailsRepository detailsRepo = new MongoRyaInstanceDetailsRepository(client, ryaInstanceName); + final RyaDetails ryaDetails = detailsRepo.getRyaInstanceDetails(); + + // Ensure PCJs are enabled. + if(!ryaDetails.getPCJIndexDetails().isEnabled()) { + throw new RyaClientException("PCJs are not enabled for the Rya instance named '" + ryaInstanceName + "'."); + } + + // Ensure the PCJ exists. + if(!ryaDetails.getPCJIndexDetails().getPCJDetails().containsKey(pcjId)) { + throw new PCJDoesNotExistException("The PCJ with id '" + pcjId + "' does not exist within Rya instance '" + ryaInstanceName + "'."); + } + + } catch(final NotInitializedException e) { + throw new InstanceDoesNotExistException("No RyaDetails are initialized for the Rya instance named '" + ryaInstanceName + "'.", e); + } catch (final RyaDetailsRepositoryException e) { + throw new RyaClientException("Could not fetch the RyaDetails for the Rya instance named '" + ryaInstanceName + "'.", e); + } + } + + private void updatePCJResults(final String ryaInstanceName, final String pcjId, final MongoClient client) throws InstanceDoesNotExistException, PCJDoesNotExistException, RyaClientException { + // Things that have to be closed before we exit. + Sail sail = null; + SailConnection sailConn = null; + + try(final PrecomputedJoinStorage pcjStorage = new MongoPcjStorage(client, ryaInstanceName)) { + // Create an instance of Sail backed by the Rya instance. + sail = connectToRya(ryaInstanceName); + final SailRepository sailRepo = new SailRepository(sail); + final SailRepositoryConnection sailRepoConn = sailRepo.getConnection(); + // Purge the old results from the PCJ. + try { + pcjStorage.purge(pcjId); + } catch (final PCJStorageException e) { + throw new RyaClientException("Could not batch update PCJ with ID '" + pcjId + "' because the old " + + "results could not be purged from it.", e); + } + + // Parse the PCJ's SPARQL query. + final PcjMetadata metadata = pcjStorage.getPcjMetadata(pcjId); + final String sparql = metadata.getSparql(); + sailConn = sail.getConnection(); + final TupleQuery tupleQuery = sailRepoConn.prepareTupleQuery(QueryLanguage.SPARQL, sparql); + + // Execute the query. + final List<VisibilityBindingSet> batch = new ArrayList<>(1000); + tupleQuery.evaluate(new TupleQueryResultHandlerBase() { + @Override + public void handleSolution(final BindingSet bindingSet) throws TupleQueryResultHandlerException { + final VisibilityBindingSet result = new VisibilityBindingSet(bindingSet, ""); + log.warn("Visibility information on the binding set is lost during a batch update." + + " This can create data leaks."); + batch.add(result); + + if(batch.size() == 1000) { + try { + pcjStorage.addResults(pcjId, batch); + } catch (final PCJStorageException e) { + throw new TupleQueryResultHandlerException("Fail to batch load new results into the PCJ with ID '" + pcjId + "'.", e); + } + batch.clear(); + } + } + }); + + if(!batch.isEmpty()) { + pcjStorage.addResults(pcjId, batch); + batch.clear(); + } + } catch(final MalformedQueryException | PCJStorageException | SailException | + QueryEvaluationException | RepositoryException | TupleQueryResultHandlerException e) { + throw new RyaClientException("Fail to batch load new results into the PCJ with ID '" + pcjId + "'.", e); + } finally { + if(sailConn != null) { + try { + sailConn.close(); + } catch (final SailException e) { + log.warn(e.getMessage(), e); + } + } + + if(sail != null) { + try { + sail.shutDown(); + } catch (final SailException e) { + log.warn(e.getMessage(), e); + } + } + } + } + + private Sail connectToRya(final String ryaInstanceName) throws RyaClientException { + try { + final MongoDBRdfConfiguration ryaConf = connectionDetails.build(ryaInstanceName); + return RyaSailFactory.getInstance(ryaConf); + } catch (SailException | AccumuloException | AccumuloSecurityException | RyaDAOException | InferenceEngineException e) { + throw new RyaClientException("Could not connect to the Rya instance named '" + ryaInstanceName + "'.", e); + } + } + + private void updatePCJMetadata(final String ryaInstanceName, final String pcjId, final MongoClient client) throws RyaClientException { + // Update the PCJ's metadata to indicate it was just batch updated. + try { + final RyaDetailsRepository detailsRepo = new MongoRyaInstanceDetailsRepository(client, ryaInstanceName); + + new RyaDetailsUpdater(detailsRepo).update(new RyaDetailsMutator() { + @Override + public RyaDetails mutate(final RyaDetails originalDetails) throws CouldNotApplyMutationException { + // Update the original PCJ Details to indicate they were batch updated. + final PCJDetails originalPCJDetails = originalDetails.getPCJIndexDetails().getPCJDetails().get(pcjId); + final PCJDetails.Builder mutatedPCJDetails = PCJDetails.builder( originalPCJDetails ) + .setUpdateStrategy( PCJUpdateStrategy.BATCH ) + .setLastUpdateTime( new Date()); + + // Replace the old PCJ Details with the updated ones. + final RyaDetails.Builder builder = RyaDetails.builder(originalDetails); + builder.getPCJIndexDetails().addPCJDetails( mutatedPCJDetails ); + return builder.build(); + } + }); + } catch (final RyaDetailsRepositoryException | CouldNotApplyMutationException e) { + throw new RyaClientException("Could not update the PCJ's metadata.", e); + } + } +} http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c826ffea/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoCreatePCJ.java ---------------------------------------------------------------------- diff --git a/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoCreatePCJ.java b/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoCreatePCJ.java new file mode 100644 index 0000000..49e7551 --- /dev/null +++ b/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoCreatePCJ.java @@ -0,0 +1,77 @@ +/** + * 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.api.client.mongo; + +import static java.util.Objects.requireNonNull; + +import java.util.Set; + +import org.apache.rya.api.client.CreatePCJ; +import org.apache.rya.api.client.InstanceDoesNotExistException; +import org.apache.rya.api.client.InstanceExists; +import org.apache.rya.api.client.RyaClientException; +import org.apache.rya.indexing.pcj.storage.PrecomputedJoinStorage.PCJStorageException; +import org.apache.rya.indexing.pcj.storage.mongo.MongoPcjStorage; + +import com.google.common.collect.Sets; +import com.mongodb.MongoClient; + +/** + * A Mongo implementation of {@link CreatePCJ}. + */ +public class MongoCreatePCJ implements CreatePCJ { + private final InstanceExists instanceExists; + private final MongoClient mongoClient; + + /** + * Constructs an instance of {@link MongoCreatePCJ}. + * + * @param mongoClient - The {@link MongoClient} used to create a new PCJ. (not null) + * @param instanceExists - The interactor used to check if a Rya instance exists. (not null) + */ + public MongoCreatePCJ( + final MongoClient mongoClient, + final MongoInstanceExists instanceExists) { + this.mongoClient = requireNonNull(mongoClient); + this.instanceExists = requireNonNull(instanceExists); + } + + @Override + public String createPCJ(final String ryaInstanceName, final String sparql, final Set<ExportStrategy> strategies) throws InstanceDoesNotExistException, RyaClientException { + requireNonNull(ryaInstanceName); + requireNonNull(sparql); + + // Ensure the Rya Instance exists. + if (!instanceExists.exists(ryaInstanceName)) { + throw new InstanceDoesNotExistException(String.format("There is no Rya instance named '%s'.", ryaInstanceName)); + } + + try(final MongoPcjStorage pcjStore = new MongoPcjStorage(mongoClient, ryaInstanceName)) { + return pcjStore.createPcj(sparql); + } catch (final PCJStorageException e) { + throw new RyaClientException("Unable to create PCJ for: " + sparql, e); + } + } + + @Override + public String createPCJ(final String instanceName, final String sparql) throws InstanceDoesNotExistException, RyaClientException { + return createPCJ(instanceName, sparql, Sets.newHashSet(ExportStrategy.RYA)); + } + +} http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c826ffea/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoDeletePCJ.java ---------------------------------------------------------------------- diff --git a/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoDeletePCJ.java b/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoDeletePCJ.java new file mode 100644 index 0000000..8c0a71d --- /dev/null +++ b/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoDeletePCJ.java @@ -0,0 +1,68 @@ +/** + * 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.api.client.mongo; + +import static java.util.Objects.requireNonNull; + +import org.apache.rya.api.client.DeletePCJ; +import org.apache.rya.api.client.InstanceDoesNotExistException; +import org.apache.rya.api.client.InstanceExists; +import org.apache.rya.api.client.RyaClientException; +import org.apache.rya.indexing.pcj.storage.PrecomputedJoinStorage.PCJStorageException; +import org.apache.rya.indexing.pcj.storage.mongo.MongoPcjStorage; + +import com.mongodb.MongoClient; + +/** + * A Mongo implementation of {@link DeletePCJ}. + */ +public class MongoDeletePCJ implements DeletePCJ { + private final InstanceExists instanceExists; + private final MongoClient mongoClient; + + /** + * Constructs an instance of {@link MongoDeletePCJ}. + * + * @param mongoClient - The {@link MongoClient} to use to delete a PCJ. (not null) + * @param instanceExists - The interactor used to check if a Rya instance exists. (not null) + */ + public MongoDeletePCJ( + final MongoClient mongoClient, + final MongoInstanceExists instanceExists) { + this.mongoClient = requireNonNull(mongoClient); + this.instanceExists = requireNonNull(instanceExists); + } + + @Override + public void deletePCJ(final String ryaInstanceName, final String pcjId) throws InstanceDoesNotExistException, RyaClientException { + requireNonNull(ryaInstanceName); + requireNonNull(pcjId); + + // Ensure the Rya Instance exists. + if (!instanceExists.exists(ryaInstanceName)) { + throw new InstanceDoesNotExistException(String.format("There is no Rya instance named '%s'.", ryaInstanceName)); + } + + try(final MongoPcjStorage pcjStore = new MongoPcjStorage(mongoClient, ryaInstanceName)) { + pcjStore.dropPcj(pcjId); + } catch (final PCJStorageException e) { + throw new RyaClientException("Unable to drop PCJ : " + pcjId, e); + } + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c826ffea/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoInstall.java ---------------------------------------------------------------------- diff --git a/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoInstall.java b/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoInstall.java index 51bd4bf..264dd78 100644 --- a/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoInstall.java +++ b/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoInstall.java @@ -107,7 +107,7 @@ public class MongoInstall implements Install { // This can only happen if somebody else installs an instance of Rya with the name between the check and now. throw new DuplicateInstanceNameException( "An instance of Rya has already been installed to this Rya storage " + - "with the name '" + instanceName + "'. Try again with a different name."); + "with the name '" + instanceName + "'. Try again with a different name."); } catch (final RyaDetailsRepositoryException e) { throw new RyaClientException("The RyaDetails couldn't be initialized. Details: " + e.getMessage(), e); } @@ -147,27 +147,28 @@ public class MongoInstall implements Install { final InstallConfiguration installConfig) throws AlreadyInitializedException, RyaDetailsRepositoryException { final RyaDetailsRepository detailsRepo = new MongoRyaInstanceDetailsRepository(adminClient, instanceName); - // Build the PCJ Index details. [not supported in mongo] - final PCJIndexDetails.Builder pcjDetailsBuilder = PCJIndexDetails.builder().setEnabled(false); + if(installConfig.getFluoPcjAppName().isPresent()) { + log.warn("Mongo does not have fluo support, use ignoring the configured fluo application name: " + installConfig.getFluoPcjAppName().get()); + } + + // Build the PCJ Index details. + final PCJIndexDetails.Builder pcjDetailsBuilder = PCJIndexDetails.builder() + .setEnabled(installConfig.isPcjIndexEnabled()); final RyaDetails details = RyaDetails.builder() // General Metadata .setRyaInstanceName(instanceName).setRyaVersion(getVersion()) - // FIXME RYA-215 .setGeoIndexDetails(new GeoIndexDetails(installConfig.isGeoIndexEnabled())) - // Secondary Index Values + // FIXME RYA-215 .setGeoIndexDetails(new GeoIndexDetails(installConfig.isGeoIndexEnabled())) .setTemporalIndexDetails(new TemporalIndexDetails(installConfig.isTemporalIndexEnabled())) - .setFreeTextDetails(new FreeTextIndexDetails(installConfig.isFreeTextIndexEnabled()))// - - // Entity centric indexing is not supported in Mongo DB. + .setFreeTextDetails(new FreeTextIndexDetails(installConfig.isFreeTextIndexEnabled())) .setEntityCentricIndexDetails(new EntityCentricIndexDetails(false)) - .setPCJIndexDetails(pcjDetailsBuilder) // Statistics values. - .setProspectorDetails(new ProspectorDetails(Optional.<Date> absent()))// - .setJoinSelectivityDetails(new JoinSelectivityDetails(Optional.<Date> absent()))// + .setProspectorDetails(new ProspectorDetails(Optional.<Date> absent())) + .setJoinSelectivityDetails(new JoinSelectivityDetails(Optional.<Date> absent())) .build(); // Initialize the table. @@ -191,26 +192,23 @@ public class MongoInstall implements Install { final MongoDBRdfConfiguration conf = connectionDetails.build(ryaDetails.getRyaInstanceName()); - // The Mongo implementation of Rya does not currently support PCJs. - if(ryaDetails.getPCJIndexDetails().isEnabled()) { - log.warn("The install configuration says to enable PCJ indexing, but Mongo RYA does not support that " + - "feature. Ignoring this configuration."); - } - conf.set(ConfigUtils.USE_PCJ, "false"); + conf.setBoolean(ConfigUtils.USE_PCJ, ryaDetails.getPCJIndexDetails().isEnabled()); // Mongo does not support entity indexing. if(ryaDetails.getEntityCentricIndexDetails().isEnabled()) { log.warn("The install configuration says to enable Entity Centric indexing, but Mongo RYA does not support " + "that feature. Ignoring this configuration."); } - conf.set(ConfigUtils.USE_ENTITY, "false"); + + //TODO mongo now has an entity index, just needs CLI support. + conf.setBoolean(ConfigUtils.USE_ENTITY, false); // FIXME RYA-215 We haven't enabled geo indexing in the console yet. //conf.set(OptionalConfigUtils.USE_GEO, "" + details.getGeoIndexDetails().isEnabled() ); // Enable the supported indexers that the instance is configured to use. - conf.set(ConfigUtils.USE_FREETEXT, "" + ryaDetails.getFreeTextIndexDetails().isEnabled()); - conf.set(ConfigUtils.USE_TEMPORAL, "" + ryaDetails.getTemporalIndexDetails().isEnabled()); + conf.setBoolean(ConfigUtils.USE_FREETEXT, ryaDetails.getFreeTextIndexDetails().isEnabled()); + conf.setBoolean(ConfigUtils.USE_TEMPORAL, ryaDetails.getTemporalIndexDetails().isEnabled()); // This initializes the living indexers that will be used by the application and // caches them within the configuration object so that they may be used later. http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c826ffea/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoRyaClientFactory.java ---------------------------------------------------------------------- diff --git a/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoRyaClientFactory.java b/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoRyaClientFactory.java index 55c6e8a..fbbec2a 100644 --- a/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoRyaClientFactory.java +++ b/extras/indexing/src/main/java/org/apache/rya/api/client/mongo/MongoRyaClientFactory.java @@ -55,12 +55,12 @@ public class MongoRyaClientFactory { return new RyaClient( new MongoInstall(connectionDetails, adminClient, instanceExists), + new MongoCreatePCJ(adminClient, instanceExists), + new MongoDeletePCJ(adminClient, instanceExists), Optional.empty(), Optional.empty(), Optional.empty(), - Optional.empty(), - Optional.empty(), - Optional.empty(), + new MongoBatchUpdatePCJ(connectionDetails, adminClient, instanceExists), new MongoGetInstanceDetails(adminClient, instanceExists), instanceExists, new MongoListInstances(adminClient), http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/c826ffea/extras/indexing/src/main/java/org/apache/rya/indexing/accumulo/ConfigUtils.java ---------------------------------------------------------------------- diff --git a/extras/indexing/src/main/java/org/apache/rya/indexing/accumulo/ConfigUtils.java b/extras/indexing/src/main/java/org/apache/rya/indexing/accumulo/ConfigUtils.java index 5cc1c44..561f6c6 100644 --- a/extras/indexing/src/main/java/org/apache/rya/indexing/accumulo/ConfigUtils.java +++ b/extras/indexing/src/main/java/org/apache/rya/indexing/accumulo/ConfigUtils.java @@ -80,25 +80,25 @@ public class ConfigUtils { */ @Deprecated public static final String CLOUDBASE_TBL_PREFIX = RdfCloudTripleStoreConfiguration.CONF_TBL_PREFIX; - + /** * @Deprecated use {@link AccumuloRdfConfiguration#CLOUDBASE_INSTANCE} instead. */ @Deprecated public static final String CLOUDBASE_INSTANCE = AccumuloRdfConfiguration.CLOUDBASE_INSTANCE; - + /** * @Deprecated use {@link AccumuloRdfConfiguration#CLOUDBASE_ZOOKEEPERS} instead. */ @Deprecated public static final String CLOUDBASE_ZOOKEEPERS = AccumuloRdfConfiguration.CLOUDBASE_ZOOKEEPERS; - + /** * @Deprecated use {@link AccumuloRdfConfiguration#CLOUDBASE_USER} instead. */ @Deprecated public static final String CLOUDBASE_USER = AccumuloRdfConfiguration.CLOUDBASE_USER; - + /** * @Deprecated use {@link AccumuloRdfConfiguration#CLOUDBASE_PASSWORD} instead. */ @@ -192,7 +192,7 @@ public class ConfigUtils { * null. Future, get table prefix from RyaDetails -- the Rya instance name * -- also getting info from the RyaDetails should happen within * RyaSailFactory and not ConfigUtils. - * + * * @param conf * Rya configuration map where it extracts the prefix (instance * name) @@ -220,7 +220,7 @@ public class ConfigUtils { /** * Used for indexing statements about date & time instances and intervals. - * + * * @param conf * @return Set of predicate URI's whose objects should be date time * literals. @@ -337,8 +337,8 @@ public class ConfigUtils { public static Instance getInstance(final Configuration conf) { // Pull out the Accumulo specific configuration values. final AccumuloRdfConfiguration accConf = new AccumuloRdfConfiguration(conf); - String instanceName = accConf.getInstanceName(); - String zoookeepers = accConf.getZookeepers(); + final String instanceName = accConf.getInstanceName(); + final String zoookeepers = accConf.getZookeepers(); // Create an Instance a mock if the mock flag is set. if (useMockInstance(conf)) { @@ -466,6 +466,10 @@ public class ConfigUtils { indexList.add(MongoTemporalIndexer.class.getName()); useFilterIndex = true; } + + if (getUsePCJ(conf) && getUseOptimalPCJ(conf)) { + conf.setPcjOptimizer(PCJOptimizer.class); + } } else { if (getUsePCJ(conf) || getUseOptimalPCJ(conf)) { conf.setPcjOptimizer(PCJOptimizer.class);
