You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rya.apache.org by dl...@apache.org on 2017/08/30 20:31:43 UTC
[05/14] incubator-rya git commit: RYA-324,
RYA-272 Geo refactoring and examples closes #182
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/9e76b8d7/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalIndexSetProvider.java
----------------------------------------------------------------------
diff --git a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalIndexSetProvider.java b/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalIndexSetProvider.java
deleted file mode 100644
index bf12f26..0000000
--- a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalIndexSetProvider.java
+++ /dev/null
@@ -1,239 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.rya.indexing.geotemporal;
-
-import static java.util.Objects.requireNonNull;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-
-import org.apache.log4j.Logger;
-import org.apache.rya.indexing.IndexingExpr;
-import org.apache.rya.indexing.IndexingFunctionRegistry;
-import org.apache.rya.indexing.IndexingFunctionRegistry.FUNCTION_TYPE;
-import org.apache.rya.indexing.accumulo.geo.GeoParseUtils;
-import org.apache.rya.indexing.accumulo.geo.GeoTupleSet;
-import org.apache.rya.indexing.external.matching.ExternalSetProvider;
-import org.apache.rya.indexing.external.matching.QuerySegment;
-import org.apache.rya.indexing.geotemporal.model.EventQueryNode;
-import org.apache.rya.indexing.geotemporal.model.EventQueryNode.EventQueryNodeBuilder;
-import org.apache.rya.indexing.geotemporal.storage.EventStorage;
-import org.openrdf.model.URI;
-import org.openrdf.model.impl.URIImpl;
-import org.openrdf.query.algebra.FunctionCall;
-import org.openrdf.query.algebra.QueryModelNode;
-import org.openrdf.query.algebra.StatementPattern;
-import org.openrdf.query.algebra.Var;
-import org.openrdf.query.algebra.helpers.QueryModelVisitorBase;
-
-import com.google.common.collect.HashMultimap;
-import com.google.common.collect.Multimap;
-
-/**
- * Provides {@link GeoTupleSet}s.
- */
-public class GeoTemporalIndexSetProvider implements ExternalSetProvider<EventQueryNode> {
- private static final Logger LOG = Logger.getLogger(GeoTemporalIndexSetProvider.class);
-
- //organzied by object var. Each object is a filter, or set of filters
- private Multimap<Var, IndexingExpr> filterMap;
-
- //organzied by subject var. Each subject is a GeoTemporalTupleSet
- private Multimap<Var, StatementPattern> patternMap;
-
- //filters that have not been constrained by statement patterns into indexing expressions yet.
- private Multimap<Var, FunctionCall> unmatchedFilters;
- //filters that have been used, to be used by the matcher later.
- private Multimap<Var, FunctionCall> matchedFilters;
-
- //organzied by object var. Used to find matches between unmatch filters and patterns
- private Map<Var, StatementPattern> objectPatterns;
-
-
- private static URI filterURI;
-
- private final EventStorage eventStorage;
-
- public GeoTemporalIndexSetProvider(final EventStorage eventStorage) {
- this.eventStorage = requireNonNull(eventStorage);
- }
-
- @Override
- public List<EventQueryNode> getExternalSets(final QuerySegment<EventQueryNode> node) {
- filterMap = HashMultimap.create();
- patternMap = HashMultimap.create();
- unmatchedFilters = HashMultimap.create();
- matchedFilters = HashMultimap.create();
-
- objectPatterns = new HashMap<>();
- //discover entities
- buildMaps(node);
- final List<EventQueryNode> nodes = createNodes();
-
- return nodes;
- }
-
- private List<EventQueryNode> createNodes() {
- final List<EventQueryNode> nodes = new ArrayList<>();
- for(final Var subj : patternMap.keySet()) {
- final EventQueryNode node = getGeoTemporalNode(subj);
- if(node != null) {
- nodes.add(node);
- }
- }
- return nodes;
- }
-
- private EventQueryNode getGeoTemporalNode(final Var subj) {
- final Collection<StatementPattern> patterns = patternMap.get(subj);
- final Collection<FunctionCall> usedFilters = new ArrayList<>();
- Optional<StatementPattern> geoPattern = Optional.empty();
- Optional<StatementPattern> temporalPattern = Optional.empty();
- Optional<Collection<IndexingExpr>> geoFilters = Optional.empty();
- Optional<Collection<IndexingExpr>> temporalFilters = Optional.empty();
-
- //should only be 2 patterns.
- for(final StatementPattern sp : patterns) {
- final Var obj = sp.getObjectVar();
-
- ///filter map does not have -const-
-
-
- if(filterMap.containsKey(obj)) {
- final Collection<IndexingExpr> filters = filterMap.get(obj);
- final IndexingFunctionRegistry.FUNCTION_TYPE type = ensureSameType(filters);
- if(type != null && type == FUNCTION_TYPE.GEO) {
- geoPattern = Optional.of(sp);
- geoFilters = Optional.of(filters);
- usedFilters.addAll(matchedFilters.get(obj));
- } else if(type != null && type == FUNCTION_TYPE.TEMPORAL) {
- temporalPattern = Optional.of(sp);
- temporalFilters = Optional.of(filters);
- usedFilters.addAll(matchedFilters.get(obj));
- } else {
- return null;
- }
- } else {
- return null;
- }
- }
-
- if(geoFilters.isPresent() && temporalFilters.isPresent() && geoPattern.isPresent() && temporalPattern.isPresent()) {
- return new EventQueryNodeBuilder()
- .setStorage(eventStorage)
- .setGeoPattern(geoPattern.get())
- .setTemporalPattern(temporalPattern.get())
- .setGeoFilters(geoFilters.get())
- .setTemporalFilters(temporalFilters.get())
- .setUsedFilters(usedFilters)
- .build();
- } else {
- return null;
- }
- }
-
- private static FUNCTION_TYPE ensureSameType(final Collection<IndexingExpr> filters) {
- FUNCTION_TYPE type = null;
- for(final IndexingExpr filter : filters) {
- if(type == null) {
- type = IndexingFunctionRegistry.getFunctionType(filter.getFunction());
- } else {
- if(IndexingFunctionRegistry.getFunctionType(filter.getFunction()) != type) {
- return null;
- }
- }
- }
- return type;
- }
-
- private void buildMaps(final QuerySegment<EventQueryNode> node) {
- final List<QueryModelNode> unused = new ArrayList<>();
- for (final QueryModelNode pattern : node.getOrderedNodes()) {
- if(pattern instanceof FunctionCall) {
- discoverFilter((FunctionCall) pattern, unused);
- }
- if(pattern instanceof StatementPattern) {
- discoverPatterns((StatementPattern) pattern, unused);
- }
- }
- }
-
- private void discoverFilter(final FunctionCall filter, final List<QueryModelNode> unmatched) {
- try {
- filter.visit(new FilterVisitor());
- } catch (final Exception e) {
- LOG.error("Failed to match the filter object.", e);
- }
- }
-
- private void discoverPatterns(final StatementPattern pattern, final List<QueryModelNode> unmatched) {
- final Var subj = pattern.getSubjectVar();
- final Var objVar = pattern.getObjectVar();
-
- patternMap.put(subj, pattern);
- objectPatterns.put(objVar, pattern);
- //check for existing filters.
- if(unmatchedFilters.containsKey(objVar)) {
- final Collection<FunctionCall> calls = unmatchedFilters.removeAll(objVar);
- for(final FunctionCall call : calls) {
- addFilter(call);
- matchedFilters.put(objVar, call);
- }
- }
- }
-
- @Override
- public Iterator<List<EventQueryNode>> getExternalSetCombos(final QuerySegment<EventQueryNode> segment) {
- final List<List<EventQueryNode>> comboList = new ArrayList<>();
- comboList.add(getExternalSets(segment));
- return comboList.iterator();
- }
-
- private void addFilter(final FunctionCall call) {
- filterURI = new URIImpl(call.getURI());
- final Var objVar = IndexingFunctionRegistry.getResultVarFromFunctionCall(filterURI, call.getArgs());
- filterMap.put(objVar, new IndexingExpr(filterURI, objectPatterns.get(objVar), GeoParseUtils.extractArguments(objVar.getName(), call)));
- }
-
- /**
- * Finds the object/function in a Filter. If the associated statement pattern
- * has been found, creates the {@link IndexingExpr} and adds it to the map.
- */
- private class FilterVisitor extends QueryModelVisitorBase<Exception> {
- @Override
- public void meet(final FunctionCall call) throws Exception {
- filterURI = new URIImpl(call.getURI());
- final FUNCTION_TYPE type = IndexingFunctionRegistry.getFunctionType(filterURI);
- if(type == FUNCTION_TYPE.GEO || type == FUNCTION_TYPE.TEMPORAL) {
- final Var objVar = IndexingFunctionRegistry.getResultVarFromFunctionCall(filterURI, call.getArgs());
- if(objectPatterns.containsKey(objVar)) {
- filterMap.put(objVar, new IndexingExpr(filterURI, objectPatterns.get(objVar), GeoParseUtils.extractArguments(objVar.getName(), call)));
- matchedFilters.put(objVar, call);
- } else {
- unmatchedFilters.put(objVar, call);
- }
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/9e76b8d7/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalIndexer.java
----------------------------------------------------------------------
diff --git a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalIndexer.java b/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalIndexer.java
deleted file mode 100644
index cbc978b..0000000
--- a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalIndexer.java
+++ /dev/null
@@ -1,193 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.rya.indexing.geotemporal;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.rya.api.persist.index.RyaSecondaryIndexer;
-import org.apache.rya.indexing.GeoConstants;
-import org.apache.rya.indexing.geotemporal.storage.EventStorage;
-import org.openrdf.model.Statement;
-import org.openrdf.model.URI;
-import org.openrdf.model.impl.URIImpl;
-
-/**
- * A repository to store, index, and retrieve {@link Statement}s based on geotemporal features.
- */
-public interface GeoTemporalIndexer extends RyaSecondaryIndexer {
-
- /**
- * Creates the {@link Eventtorage} that will be used by the indexer.
- *
- * @param conf - Indicates how the {@link EventStorage} is initialized. (not null)
- * @return The {@link EventStorage} that will be used by this indexer.
- */
- public abstract EventStorage getEventStorage(final Configuration conf);
-
- /**
- * Used to indicate which geo filter functions to use in a query.
- */
- public static enum GeoPolicy {
- /**
- * The provided geo object equals the geo object where the event took place.
- */
- EQUALS(GeoConstants.GEO_SF_EQUALS),
-
- /**
- * The provided geo object does not share any space with the event.
- */
- DISJOINT(GeoConstants.GEO_SF_DISJOINT),
-
- /**
- * The provided geo object shares some amount of space with the event.
- */
- INTERSECTS(GeoConstants.GEO_SF_INTERSECTS),
-
- /**
- * The provided geo object shares a point with the event, but only on the edge.
- */
- TOUCHES(GeoConstants.GEO_SF_TOUCHES),
-
- /**
- * The provided geo object shares some, but not all space with the event.
- */
- CROSSES(GeoConstants.GEO_SF_CROSSES),
-
- /**
- * The provided geo object exists completely within the event.
- */
- WITHIN(GeoConstants.GEO_SF_WITHIN),
-
- /**
- * The event took place completely within the provided geo object.
- */
- CONTAINS(GeoConstants.GEO_SF_CONTAINS),
-
- /**
- * The provided geo object has some but not all points in common with the event,
- * are of the same dimension, and the intersection of the interiors has the
- * same dimension as the geometries themselves.
- */
- OVERLAPS(GeoConstants.GEO_SF_OVERLAPS);
-
- private final URI uri;
-
- private GeoPolicy(final URI uri) {
- this.uri = uri;
- }
-
- public URI getURI() {
- return uri;
- }
-
- public static GeoPolicy fromURI(final URI uri) {
- for(final GeoPolicy policy : GeoPolicy.values()) {
- if(policy.getURI().equals(uri)) {
- return policy;
- }
- }
- return null;
- }
- }
-
- static final String TEMPORAL_NS = "tag:rya-rdf.org,2015:temporal#";
- /**
- * Used to indicate which temporal filter functions to use in a query.
- */
- public enum TemporalPolicy {
- /**
- * The provided instant in time equals the instant the event took place.
- */
- INSTANT_EQUALS_INSTANT(true, new URIImpl(TEMPORAL_NS+"equals")),
-
- /**
- * The provided instant in time was before when the event took place.
- */
- INSTANT_BEFORE_INSTANT(true, new URIImpl(TEMPORAL_NS+"before")),
-
- /**
- * The provided instant in time was after when the event took place.
- */
- INSTANT_AFTER_INSTANT(true, new URIImpl(TEMPORAL_NS+"after")),
-
- /**
- * The provided instant in time was before a time period.
- */
- INSTANT_BEFORE_INTERVAL(false, new URIImpl(TEMPORAL_NS+"beforeInterval")),
-
- /**
- * The provided instant in time took place within a set of time.
- */
- INSTANT_IN_INTERVAL(false, new URIImpl(TEMPORAL_NS+"insideInterval")),
-
- /**
- * The provided instant in time took place after a time period.
- */
- INSTANT_AFTER_INTERVAL(false, new URIImpl(TEMPORAL_NS+"afterInterval")),
-
- /**
- * The provided instant in time equals the start of the interval in which the event took place.
- */
- INSTANT_START_INTERVAL(false, new URIImpl(TEMPORAL_NS+"hasBeginningInterval")),
-
- /**
- * The provided instant in time equals the end of the interval in which the event took place.
- */
- INSTANT_END_INTERVAL(false, new URIImpl(TEMPORAL_NS+"hasEndInterval")),
-
- /**
- * The provided interval equals the interval in which the event took place.
- */
- INTERVAL_EQUALS(false, new URIImpl(TEMPORAL_NS+"intervalEquals")),
-
- /**
- * The provided interval is before the interval in which the event took place.
- */
- INTERVAL_BEFORE(false, new URIImpl(TEMPORAL_NS+"intervalBefore")),
-
- /**
- * The provided interval is after the interval in which the event took place.
- */
- INTERVAL_AFTER(false, new URIImpl(TEMPORAL_NS+"intervalAfter"));
-
- private final boolean isInstant;
- private final URI uri;
-
- TemporalPolicy(final boolean isInstant, final URI uri) {
- this.isInstant = isInstant;
- this.uri = uri;
- }
-
- public boolean isInstant(){
- return isInstant;
- }
-
- public URI getURI() {
- return uri;
- }
-
- public static TemporalPolicy fromURI(final URI uri) {
- for(final TemporalPolicy policy : TemporalPolicy.values()) {
- if(policy.getURI().equals(uri)) {
- return policy;
- }
- }
- return null;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/9e76b8d7/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalIndexerFactory.java
----------------------------------------------------------------------
diff --git a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalIndexerFactory.java b/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalIndexerFactory.java
deleted file mode 100644
index a7ba8fa..0000000
--- a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalIndexerFactory.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.rya.indexing.geotemporal;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.rya.indexing.accumulo.ConfigUtils;
-import org.apache.rya.indexing.geotemporal.mongo.MongoGeoTemporalIndexer;
-import org.apache.rya.mongodb.MongoDBRdfConfiguration;
-import org.apache.rya.mongodb.MongoSecondaryIndex;
-
-/**
- * Factory for retrieving a {@link GeoTemporalIndexer} based on a provided {@link Configuration}.
- */
-public class GeoTemporalIndexerFactory {
- /**
- * Creates and returns a {@link GeoTemporalIndexer}.
- * @param conf - The {@link Configuration} to base the {@link GeoTemporalIndexer} on.
- * @return The created {@link GeoTemporalIndexer}.
- */
- public GeoTemporalIndexer getIndexer(final Configuration conf) {
- if(ConfigUtils.getUseMongo(conf)) {
- final MongoDBRdfConfiguration config = new MongoDBRdfConfiguration(conf);
- for(final MongoSecondaryIndex index : config.getAdditionalIndexers()) {
- if(index instanceof GeoTemporalIndexer) {
- return (GeoTemporalIndexer) index;
- }
- }
- final MongoGeoTemporalIndexer index = new MongoGeoTemporalIndexer();
- index.setConf(conf);
- index.init();
- return index;
- } else {
- //TODO: add Accumulo here.
- return null;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/9e76b8d7/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalOptimizer.java
----------------------------------------------------------------------
diff --git a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalOptimizer.java b/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalOptimizer.java
deleted file mode 100644
index d626adc..0000000
--- a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalOptimizer.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.rya.indexing.geotemporal;
-
-import org.apache.hadoop.conf.Configurable;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.rya.indexing.external.matching.AbstractExternalSetOptimizer;
-import org.apache.rya.indexing.external.matching.ExternalSetMatcher;
-import org.apache.rya.indexing.external.matching.ExternalSetProvider;
-import org.apache.rya.indexing.external.matching.QueryNodeListRater;
-import org.apache.rya.indexing.external.matching.QuerySegment;
-import org.apache.rya.indexing.geotemporal.model.EventQueryNode;
-
-import com.google.common.base.Optional;
-
-
-public class GeoTemporalOptimizer extends AbstractExternalSetOptimizer<EventQueryNode> implements Configurable {
- private static final GeoTemporalExternalSetMatcherFactory MATCHER_FACTORY = new GeoTemporalExternalSetMatcherFactory();
-
- private GeoTemporalIndexer indexer;
- private GeoTemporalIndexSetProvider provider;
- private Configuration conf;
-
- @Override
- public void setConf(final Configuration conf) {
- this.conf = conf;
- final GeoTemporalIndexerFactory factory = new GeoTemporalIndexerFactory();
- indexer = factory.getIndexer(conf);
-
- //conf here does not matter since EventStorage has already been set in the indexer.
- provider = new GeoTemporalIndexSetProvider(indexer.getEventStorage(conf));
- }
-
- @Override
- public Configuration getConf() {
- return conf;
- }
-
- @Override
- protected ExternalSetMatcher<EventQueryNode> getMatcher(final QuerySegment<EventQueryNode> segment) {
- return MATCHER_FACTORY.getMatcher(segment);
- }
-
- @Override
- protected ExternalSetProvider<EventQueryNode> getProvider() {
- return provider;
- }
-
- @Override
- protected Optional<QueryNodeListRater> getNodeListRater(final QuerySegment<EventQueryNode> segment) {
- return null;
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/9e76b8d7/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalToSegmentConverter.java
----------------------------------------------------------------------
diff --git a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalToSegmentConverter.java b/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalToSegmentConverter.java
deleted file mode 100644
index 22bfdb1..0000000
--- a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/GeoTemporalToSegmentConverter.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.rya.indexing.geotemporal;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-
-import org.apache.rya.indexing.external.matching.ExternalSetConverter;
-import org.apache.rya.indexing.external.matching.JoinSegment;
-import org.apache.rya.indexing.external.matching.QuerySegment;
-import org.apache.rya.indexing.geotemporal.model.EventQueryNode;
-import org.openrdf.query.algebra.Filter;
-import org.openrdf.query.algebra.QueryModelNode;
-import org.openrdf.query.algebra.ValueExpr;
-
-import com.google.common.base.Preconditions;
-
-/**
- * Implementation of {@link ExternalSetConverter} to convert {@link EventQueryNode}s
- * to {@link QuerySegment}s.
- *
- */
-public class GeoTemporalToSegmentConverter implements ExternalSetConverter<EventQueryNode> {
- @Override
- public QuerySegment<EventQueryNode> setToSegment(final EventQueryNode set) {
- Preconditions.checkNotNull(set);
- final Set<QueryModelNode> matched = new HashSet<>(set.getPatterns());
- matched.addAll(set.getFilters());
- final List<QueryModelNode> unmatched = new ArrayList<>(set.getPatterns());
- return new JoinSegment<EventQueryNode>(matched, unmatched, new HashMap<ValueExpr, Filter>());
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/9e76b8d7/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/model/Event.java
----------------------------------------------------------------------
diff --git a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/model/Event.java b/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/model/Event.java
deleted file mode 100644
index 4c50bfb..0000000
--- a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/model/Event.java
+++ /dev/null
@@ -1,218 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.rya.indexing.geotemporal.model;
-
-import static java.util.Objects.requireNonNull;
-
-import java.util.Objects;
-import java.util.Optional;
-
-import org.apache.rya.api.domain.RyaURI;
-import org.apache.rya.indexing.TemporalInstant;
-import org.apache.rya.indexing.TemporalInterval;
-import org.apache.rya.indexing.geotemporal.GeoTemporalIndexer;
-
-import com.vividsolutions.jts.geom.Geometry;
-
-import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
-import edu.umd.cs.findbugs.annotations.NonNull;
-
-/**
- * Query object for a {@link GeoTemporalIndexer}.
- * Defines a {@link Geometry}, either a {@link TemporalInstant} or
- * {@link TemporalInterval}, and a triple Subject.
- */
-public class Event {
- private final Optional<Geometry> geometry;
- private final Optional<TemporalInstant> instant;
- private final Optional<TemporalInterval> interval;
- private final RyaURI subject;
-
- private final boolean isInstant;
-
- /**
- * Creates a new {@link Event} query object with a {@link TemporalInstant}.
- * @param geo - The {@link Geometry} to use when querying.
- * @param instant - The {@link TemporalInstant} to use when querying.
- * @param subject - The Subject that both statements must have when querying.
- */
- private Event(final Geometry geo, final TemporalInstant instant, final RyaURI subject) {
- this.subject = requireNonNull(subject);
-
- //these fields are nullable since they are filled field by field.
- this.instant = Optional.ofNullable(instant);
- geometry = Optional.ofNullable(geo);
- isInstant = true;
- interval = Optional.empty();
- }
-
- /**
- * Creates a new {@link Event} query object with a {@link TemporalInterval}.
- * @param geo - The {@link Geometry} to use when querying.
- * @param interval - The {@link TemporalInterval} to use when querying.
- * @param subject - The Subject that both statements must have when querying.
- */
- private Event(final Geometry geo, final TemporalInterval interval, final RyaURI subject) {
- this.subject = requireNonNull(subject);
-
- //these fields are nullable since they are filled field by field.
- this.interval = Optional.ofNullable(interval);
- geometry = Optional.ofNullable(geo);
- isInstant = false;
- instant = Optional.empty();
- }
-
- /**
- * @return Whether or not the query object uses a {@link TemporalInstant}.
- */
- public boolean isInstant() {
- return isInstant;
- }
-
- /**
- * @return The {@link Geometry} to use when querying.
- */
- public Optional<Geometry> getGeometry() {
- return geometry;
- }
-
- /**
- * @return The {@link TemporalInstant} to use when querying.
- */
- public Optional<TemporalInstant> getInstant() {
- return instant;
- }
-
- /**
- * @return The {@link TemporalInterval} to use when querying.
- */
- public Optional<TemporalInterval> getInterval() {
- return interval;
- }
-
- /**
- * @return The statement subject.
- */
- public RyaURI getSubject() {
- return subject;
- }
-
- @Override
- public int hashCode() {
- if(isInstant) {
- return Objects.hash(subject, geometry, instant);
- } else {
- return Objects.hash(subject, geometry, interval);
- }
- }
-
- @Override
- public boolean equals(final Object o) {
- if(this == o) {
- return true;
- }
- if(o instanceof Event) {
- final Event event = (Event) o;
- return Objects.equals(subject, event.subject) &&
- Objects.equals(isInstant, event.isInstant) &&
- (isInstant ? Objects.equals(instant, event.instant) : Objects.equals(interval, event.interval));
- }
- return false;
- }
-
- public static Builder builder(final Event event) {
- final Builder builder = new Builder()
- .setSubject(event.getSubject());
- if(event.getGeometry().isPresent()) {
- builder.setGeometry(event.getGeometry().get());
- }
- if(event.isInstant()) {
- if(event.getInstant().isPresent()) {
- builder.setTemporalInstant(event.getInstant().get());
- }
- } else {
- if(event.getInterval().isPresent()) {
- builder.setTemporalInterval(event.getInterval().get());
- }
- }
- return builder;
- }
-
- public static Builder builder() {
- return new Builder();
- }
-
- /**
- * Builds instances of {@link Event}.
- */
- @DefaultAnnotation(NonNull.class)
- public static class Builder {
- private RyaURI subject;
- private Geometry geo;
- private TemporalInstant instant;
- private TemporalInterval interval;
-
- /**
- * Sets the {@link RyaURI} subject.
- * @param subject - The subject to key on the event.
- */
- public Builder setSubject(final RyaURI subject) {
- this.subject = subject;
- return this;
- }
-
- /**
- * Sets the {@link Geometry}.
- * @param geo - The geometry.
- */
- public Builder setGeometry(final Geometry geo) {
- this.geo = geo;
- return this;
- }
-
- /**
- * Sets the {@link TemporalInterval}.
- * @param interval - The interval.
- */
- public Builder setTemporalInterval(final TemporalInterval interval) {
- this.interval = interval;
- return this;
- }
-
- /**
- * Sets the {@link TemporalInstant}.
- * @param instant - The instant.
- */
- public Builder setTemporalInstant(final TemporalInstant instant) {
- this.instant = instant;
- return this;
- }
-
- /**
- * @return The new {@link Event}.
- */
- public Event build() {
- if(instant == null) {
- return new Event(geo, interval, subject);
- } else {
- return new Event(geo, instant, subject);
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/9e76b8d7/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/model/EventQueryNode.java
----------------------------------------------------------------------
diff --git a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/model/EventQueryNode.java b/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/model/EventQueryNode.java
deleted file mode 100644
index 104fca8..0000000
--- a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/model/EventQueryNode.java
+++ /dev/null
@@ -1,372 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.rya.indexing.geotemporal.model;
-
-import static java.util.Objects.requireNonNull;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.Set;
-
-import org.apache.commons.lang.builder.EqualsBuilder;
-import org.apache.rya.api.domain.RyaURI;
-import org.apache.rya.indexing.IndexingExpr;
-import org.apache.rya.indexing.TemporalInstant;
-import org.apache.rya.indexing.TemporalInstantRfc3339;
-import org.apache.rya.indexing.entity.query.EntityQueryNode;
-import org.apache.rya.indexing.geotemporal.storage.EventStorage;
-import org.apache.rya.indexing.mongodb.update.RyaObjectStorage.ObjectStorageException;
-import org.apache.rya.rdftriplestore.evaluation.ExternalBatchingIterator;
-import org.joda.time.DateTime;
-import org.joda.time.DateTimeZone;
-import org.openrdf.model.Value;
-import org.openrdf.model.impl.ValueFactoryImpl;
-import org.openrdf.query.BindingSet;
-import org.openrdf.query.QueryEvaluationException;
-import org.openrdf.query.algebra.FunctionCall;
-import org.openrdf.query.algebra.StatementPattern;
-import org.openrdf.query.algebra.Var;
-import org.openrdf.query.algebra.evaluation.impl.ExternalSet;
-import org.openrdf.query.algebra.evaluation.iterator.CollectionIteration;
-import org.openrdf.query.impl.MapBindingSet;
-
-import com.vividsolutions.jts.geom.Geometry;
-
-import info.aduna.iteration.CloseableIteration;
-
-public class EventQueryNode extends ExternalSet implements ExternalBatchingIterator {
- private final Collection<FunctionCall> usedFilters;
- private final Collection<IndexingExpr> geoFilters;
- private final Collection<IndexingExpr> temporalFilters;
-
- private final StatementPattern geoPattern;
- private final StatementPattern temporalPattern;
-
- //Information about the subject of the patterns.
- private final boolean subjectIsConstant;
- private final Optional<String> subjectVar;
- //not final because if the subject is a variable and the evaluate() is
- // provided a binding set that contains the subject, this optional is used.
- private Optional<String> subjectConstant;
-
- //since and EventQueryNode exists in a single segment, all binding names are garunteed to be assured.
- private final Set<String> bindingNames;
-
- private Collection<StatementPattern> patterns;
-
- private final EventStorage eventStore;
-
- /**
- * Constructs an instance of {@link EventQueryNode}.
- * @param usedFilters
- *
- * @param type - The type of {@link Event} this node matches. (not null)
- * @param patterns - The query StatementPatterns that are solved using an
- * Event of the Type. (not null)
- * @param entities - The {@link EventStorage} that will be searched to match
- * {@link BindingSet}s when evaluating a query. (not null)
- */
- private EventQueryNode(final EventStorage eventStore, final StatementPattern geoPattern, final StatementPattern temporalPattern, final Collection<IndexingExpr> geoFilters, final Collection<IndexingExpr> temporalFilters, final Collection<FunctionCall> usedFilters) throws IllegalStateException {
- this.geoPattern = requireNonNull(geoPattern);
- this.temporalPattern = requireNonNull(temporalPattern);
- this.geoFilters = requireNonNull(geoFilters);
- this.temporalFilters = requireNonNull(temporalFilters);
- this.eventStore = requireNonNull(eventStore);
- this.usedFilters = requireNonNull(usedFilters);
- bindingNames = new HashSet<>();
-
- // Subject based preconditions.
- verifySameSubjects(getPatterns());
- // Predicate based preconditions.
- verifyAllPredicatesAreConstants(getPatterns());
-
- // The Subject may either be constant or a variable.
- final Var subject = patterns.iterator().next().getSubjectVar();
- subjectIsConstant = subject.isConstant();
- if(subjectIsConstant) {
- subjectConstant = Optional.of( subject.getValue().toString() );
- subjectVar = Optional.empty();
- } else {
- subjectConstant = Optional.empty();
- subjectVar = Optional.of( subject.getName() );
- }
- }
-
- @Override
- public Set<String> getBindingNames() {
- return bindingNames;
- }
-
- @Override
- public Set<String> getAssuredBindingNames() {
- return bindingNames;
- }
-
- /**
- * Verify the Subject for all of the patterns is the same.
- *
- * @param patterns - The patterns to check.
- * @throws IllegalStateException If all of the Subjects are not the same.
- */
- private static void verifySameSubjects(final Collection<StatementPattern> patterns) throws IllegalStateException {
- requireNonNull(patterns);
-
- final Iterator<StatementPattern> it = patterns.iterator();
- final Var subject = it.next().getSubjectVar();
-
- while(it.hasNext()) {
- final StatementPattern pattern = it.next();
- if(!pattern.getSubjectVar().equals(subject)) {
- throw new IllegalStateException("At least one of the patterns has a different subject from the others. " +
- "All subjects must be the same.");
- }
- }
- }
-
- /**
- * Verifies all of the Statement Patterns have Constants for their predicates.
- *
- * @param patterns - The patterns to check. (not null)
- * @throws IllegalStateException A pattern has a variable predicate.
- */
- private static void verifyAllPredicatesAreConstants(final Collection<StatementPattern> patterns) throws IllegalStateException {
- requireNonNull(patterns);
-
- for(final StatementPattern pattern : patterns) {
- if(!pattern.getPredicateVar().isConstant()) {
- throw new IllegalStateException("The Predicate of a Statement Pattern must be constant. Pattern: " + pattern);
- }
- }
- }
-
- @Override
- public CloseableIteration<BindingSet, QueryEvaluationException> evaluate(final BindingSet bindings) throws QueryEvaluationException {
- final List<BindingSet> list = new ArrayList<>();
- try {
- final Collection<Event> searchEvents;
- final String subj;
- //if the provided binding set has the subject already, set it to the constant subject.
- if(!subjectConstant.isPresent() && bindings.hasBinding(subjectVar.get())) {
- subjectConstant = Optional.of(bindings.getValue(subjectVar.get()).stringValue());
- } else if(bindings.size() != 0) {
- list.add(bindings);
- }
-
- // If the subject needs to be filled in, check if the subject variable is in the binding set.
- if(subjectConstant.isPresent()) {
- // if it is, fetch that value and then fetch the entity for the subject.
- subj = subjectConstant.get();
- searchEvents = eventStore.search(Optional.of(new RyaURI(subj)), Optional.of(geoFilters), Optional.of(temporalFilters));
- } else {
- searchEvents = eventStore.search(Optional.empty(), Optional.of(geoFilters), Optional.of(temporalFilters));
- }
-
- for(final Event event : searchEvents) {
- final MapBindingSet resultSet = new MapBindingSet();
- if(event.getGeometry().isPresent()) {
- final Geometry geo = event.getGeometry().get();
- final Value geoValue = ValueFactoryImpl.getInstance().createLiteral(geo.toText());
- final Var geoObj = geoPattern.getObjectVar();
- resultSet.addBinding(geoObj.getName(), geoValue);
- }
-
- final Value temporalValue;
- if(event.isInstant() && event.getInstant().isPresent()) {
- final Optional<TemporalInstant> opt = event.getInstant();
- DateTime dt = opt.get().getAsDateTime();
- dt = dt.toDateTime(DateTimeZone.UTC);
- final String str = dt.toString(TemporalInstantRfc3339.FORMATTER);
- temporalValue = ValueFactoryImpl.getInstance().createLiteral(str);
- } else if(event.getInterval().isPresent()) {
- temporalValue = ValueFactoryImpl.getInstance().createLiteral(event.getInterval().get().getAsPair());
- } else {
- temporalValue = null;
- }
-
- if(temporalValue != null) {
- final Var temporalObj = temporalPattern.getObjectVar();
- resultSet.addBinding(temporalObj.getName(), temporalValue);
- }
- list.add(resultSet);
- }
- } catch (final ObjectStorageException e) {
- throw new QueryEvaluationException("Failed to evaluate the binding set", e);
- }
- return new CollectionIteration<>(list);
- }
-
- public Collection<IndexingExpr> getGeoFilters() {
- return geoFilters;
- }
-
- public Collection<IndexingExpr> getTemporalFilters() {
- return temporalFilters;
- }
-
- public Collection<FunctionCall> getFilters() {
- return usedFilters;
- }
-
- public Collection<StatementPattern> getPatterns() {
- if(patterns == null) {
- patterns = new ArrayList<>();
- patterns.add(geoPattern);
- patterns.add(temporalPattern);
- }
- return patterns;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(subjectIsConstant,
- subjectVar,
- geoFilters,
- temporalFilters,
- geoPattern,
- temporalPattern,
- bindingNames,
- eventStore);
- }
-
- @Override
- public boolean equals(final Object other) {
- if(other instanceof EventQueryNode) {
- final EventQueryNode otherNode = (EventQueryNode)other;
- return new EqualsBuilder()
- .append(subjectIsConstant, otherNode.subjectIsConstant)
- .append(subjectVar, otherNode.subjectVar)
- .append(geoFilters, otherNode.geoFilters)
- .append(geoPattern, otherNode.geoPattern)
- .append(temporalFilters, otherNode.temporalFilters)
- .append(temporalPattern, otherNode.temporalPattern)
- .append(bindingNames, otherNode.bindingNames)
- .append(subjectConstant, otherNode.subjectConstant)
- .isEquals();
- }
- return false;
- }
-
- @Override
- public EventQueryNode clone() {
- return new EventQueryNode(eventStore, geoPattern, temporalPattern, geoFilters, temporalFilters, usedFilters);
- }
-
- @Override
- public String toString() {
- final StringBuilder sb = new StringBuilder();
- sb.append("Geo Pattern: " + geoPattern.toString());
- sb.append("\n--Geo Filters--\n");
- for(final IndexingExpr filter : geoFilters) {
- sb.append(filter.toString());
- sb.append("\n");
- }
- sb.append("\n-------------------\n");
- sb.append("Temporal Pattern: " + temporalPattern.toString());
- sb.append("\n--Temporal Filters--\n");
- for(final IndexingExpr filter : temporalFilters) {
- sb.append(filter.toString());
- sb.append("\n");
- }
- return sb.toString();
- }
-
- @Override
- public CloseableIteration<BindingSet, QueryEvaluationException> evaluate(final Collection<BindingSet> bindingset)
- throws QueryEvaluationException {
- return null;
- }
-
- /**
- * Builder for {@link EventQueryNode}s.
- */
- public static class EventQueryNodeBuilder {
- private EventStorage store;
- private StatementPattern geoPattern;
- private StatementPattern temporalPattern;
- private Collection<IndexingExpr> geoFilters;
- private Collection<IndexingExpr> temporalFilters;
- private Collection<FunctionCall> usedFilters;
-
- /**
- * @param store - The {@link EventStorage} to use in the {@link EntityQueryNode}
- * @return - The Builder.
- */
- public EventQueryNodeBuilder setStorage(final EventStorage store) {
- this.store = store;
- return this;
- }
-
- /**
- * @param geoPattern - The geo {@link StatementPattern} to use in the {@link EntityQueryNode}
- * @return - The Builder.
- */
- public EventQueryNodeBuilder setGeoPattern(final StatementPattern geoPattern) {
- this.geoPattern = geoPattern;
- return this;
- }
-
- /**
- * @param temporalPattern - The temporal {@link StatementPattern} to use in the {@link EntityQueryNode}
- * @return - The Builder.
- */
- public EventQueryNodeBuilder setTemporalPattern(final StatementPattern temporalPattern) {
- this.temporalPattern = temporalPattern;
- return this;
- }
-
- /**
- * @param geoFilters - The geo filter(s) {@link IndexingExpr} to use in the {@link EntityQueryNode}
- * @return - The Builder.
- */
- public EventQueryNodeBuilder setGeoFilters(final Collection<IndexingExpr> geoFilters) {
- this.geoFilters = geoFilters;
- return this;
- }
-
- /**
- * @param temporalFilters - The temporal filter(s) {@link IndexingExpr} to use in the {@link EntityQueryNode}
- * @return - The Builder.
- */
- public EventQueryNodeBuilder setTemporalFilters(final Collection<IndexingExpr> temporalFilters) {
- this.temporalFilters = temporalFilters;
- return this;
- }
-
- /**
- * @param usedFilters - The filter(s) used by the {@link EntityQueryNode}
- * @return - The Builder.
- */
- public EventQueryNodeBuilder setUsedFilters(final Collection<FunctionCall> usedFilters) {
- this.usedFilters = usedFilters;
- return this;
- }
-
- /**
- * @return The {@link EntityQueryNode} built by the builder.
- */
- public EventQueryNode build() {
- return new EventQueryNode(store, geoPattern, temporalPattern, geoFilters, temporalFilters, usedFilters);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/9e76b8d7/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/mongo/EventDocumentConverter.java
----------------------------------------------------------------------
diff --git a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/mongo/EventDocumentConverter.java b/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/mongo/EventDocumentConverter.java
deleted file mode 100644
index 926f357..0000000
--- a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/mongo/EventDocumentConverter.java
+++ /dev/null
@@ -1,171 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.rya.indexing.geotemporal.mongo;
-
-import static java.util.Objects.requireNonNull;
-
-import java.util.Date;
-import java.util.List;
-
-import org.apache.rya.api.domain.RyaURI;
-import org.apache.rya.indexing.TemporalInstant;
-import org.apache.rya.indexing.TemporalInstantRfc3339;
-import org.apache.rya.indexing.TemporalInterval;
-import org.apache.rya.indexing.entity.storage.mongo.DocumentConverter;
-import org.apache.rya.indexing.geotemporal.model.Event;
-import org.apache.rya.indexing.mongodb.geo.GeoMongoDBStorageStrategy;
-import org.bson.Document;
-import org.joda.time.DateTime;
-
-import com.vividsolutions.jts.geom.Coordinate;
-import com.vividsolutions.jts.geom.CoordinateList;
-import com.vividsolutions.jts.geom.Geometry;
-import com.vividsolutions.jts.geom.GeometryFactory;
-import com.vividsolutions.jts.geom.LinearRing;
-
-public class EventDocumentConverter implements DocumentConverter<Event>{
- public static final String SUBJECT = "_id";
- public static final String GEO_KEY = "location";
- public static final String INTERVAL_START = "start";
- public static final String INTERVAL_END = "end";
- public static final String INSTANT = "instant";
-
- private final GeoMongoDBStorageStrategy geoAdapter = new GeoMongoDBStorageStrategy(0.0);
-
- @Override
- public Document toDocument(final Event event) {
- requireNonNull(event);
-
- final Document doc = new Document();
- doc.append(SUBJECT, event.getSubject().getData());
-
- if(event.getGeometry().isPresent()) {
- if (event.getGeometry().get().getNumPoints() > 1) {
- doc.append(GEO_KEY, geoAdapter.getCorrespondingPoints(event.getGeometry().get()));
- } else {
- doc.append(GEO_KEY, geoAdapter.getDBPoint(event.getGeometry().get()));
- }
- }
- if(event.isInstant()) {
- if(event.getInstant().isPresent()) {
- doc.append(INSTANT, event.getInstant().get().getAsDateTime().toDate());
- }
- } else {
- if(event.getInterval().isPresent()) {
- doc.append(INTERVAL_START, event.getInterval().get().getHasBeginning().getAsDateTime().toDate());
- doc.append(INTERVAL_END, event.getInterval().get().getHasEnd().getAsDateTime().toDate());
- }
- }
-
- return doc;
- }
-
- @Override
- public Event fromDocument(final Document document) throws DocumentConverterException {
- requireNonNull(document);
-
- final boolean isInstant;
-
- // Preconditions.
- if(!document.containsKey(SUBJECT)) {
- throw new DocumentConverterException("Could not convert document '" + document +
- "' because its '" + SUBJECT + "' field is missing.");
- }
-
- if(document.containsKey(INSTANT)) {
- isInstant = true;
- } else {
- isInstant = false;
- }
-
- final String subject = document.getString(SUBJECT);
-
- final Event.Builder builder = new Event.Builder()
- .setSubject(new RyaURI(subject));
-
- if(document.containsKey(GEO_KEY)) {
- final Document geoObj = (Document) document.get(GEO_KEY);
- final GeometryFactory geoFact = new GeometryFactory();
- final String typeString = (String) geoObj.get("type");
- final CoordinateList coords = new CoordinateList();
- final Geometry geo;
- if (typeString.equals("Point")) {
- final List<Double> point = (List<Double>) geoObj.get("coordinates");
- final Coordinate coord = new Coordinate(point.get(0), point.get(1));
- geo = geoFact.createPoint(coord);
- } else if (typeString.equals("LineString")) {
- final List<List<Double>> pointsList = (List<List<Double>>) geoObj.get("coordinates");
- for (final List<Double> point : pointsList) {
- coords.add(new Coordinate(point.get(0), point.get(1)));
- }
- geo = geoFact.createLineString(coords.toCoordinateArray());
- } else {
- final List<List<List<Double>>> pointsList = (List<List<List<Double>>>) geoObj.get("coordinates");
- if(pointsList.size() == 1) {
- final List<List<Double>> poly = pointsList.get(0);
- for (final List<Double> point : poly) {
- coords.add(new Coordinate(point.get(0), point.get(1)));
- }
- geo = geoFact.createPolygon(coords.toCoordinateArray());
- } else {
- final List<List<Double>> first = pointsList.get(0);
- final CoordinateList shellCoords = new CoordinateList();
- for (final List<Double> point : pointsList.get(0)) {
- shellCoords.add(new Coordinate(point.get(0), point.get(1)));
- }
- final LinearRing shell = geoFact.createLinearRing(shellCoords.toCoordinateArray());
-
- final List<List<List<Double>>> holesPoints = pointsList.subList(1, pointsList.size() - 1);
- final LinearRing[] holes = new LinearRing[holesPoints.size()];
- for(int ii = 0; ii < holes.length; ii++) {
- final List<List<Double>> holePoints = holesPoints.get(ii);
- final CoordinateList shells = new CoordinateList();
- for (final List<Double> point : pointsList.get(0)) {
- shells.add(new Coordinate(point.get(0), point.get(1)));
- }
- holes[ii] = geoFact.createLinearRing(shells.toCoordinateArray());
- }
- geo = geoFact.createPolygon(shell,
- holes);
- }
- }
- builder.setGeometry(geo);
- }
-
- if(isInstant) {
- //we already know the key exists
- final Date date = (Date) document.get(INSTANT);
- final DateTime dt = new DateTime(date.getTime());
- final TemporalInstant instant = new TemporalInstantRfc3339(dt);
- builder.setTemporalInstant(instant);
- } else if(document.containsKey(INTERVAL_START)){
- Date date = (Date) document.get(INTERVAL_START);
- DateTime dt = new DateTime(date.getTime());
- final TemporalInstant begining = new TemporalInstantRfc3339(dt);
-
- date = (Date) document.get(INTERVAL_END);
- dt = new DateTime(date.getTime());
- final TemporalInstant end = new TemporalInstantRfc3339(dt);
-
- final TemporalInterval interval = new TemporalInterval(begining, end);
- builder.setTemporalInterval(interval);
- }
- return builder.build();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/9e76b8d7/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/mongo/EventUpdater.java
----------------------------------------------------------------------
diff --git a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/mongo/EventUpdater.java b/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/mongo/EventUpdater.java
deleted file mode 100644
index 1c62407..0000000
--- a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/mongo/EventUpdater.java
+++ /dev/null
@@ -1,85 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.rya.indexing.geotemporal.mongo;
-
-import static java.util.Objects.requireNonNull;
-
-import java.util.Optional;
-
-import org.apache.rya.api.domain.RyaURI;
-import org.apache.rya.indexing.geotemporal.model.Event;
-import org.apache.rya.indexing.geotemporal.storage.EventStorage;
-import org.apache.rya.indexing.geotemporal.storage.EventStorage.EventStorageException;
-import org.apache.rya.indexing.mongodb.update.MongoDocumentUpdater;
-import org.apache.rya.indexing.mongodb.update.RyaObjectStorage.ObjectStorageException;
-
-import edu.umd.cs.findbugs.annotations.DefaultAnnotation;
-import edu.umd.cs.findbugs.annotations.NonNull;
-
-/**
- * Performs update operations over an {@link EventStorage}.
- */
-@DefaultAnnotation(NonNull.class)
-public class EventUpdater implements MongoDocumentUpdater<RyaURI, Event>{
- private final EventStorage events;
-
- /**
- * Constructs an instance of {@link EventUpdater}
- *
- * @param events - The storage this updater operates over. (not null)
- */
- public EventUpdater(final EventStorage events) {
- this.events = requireNonNull(events);
- }
-
- @Override
- public Optional<Event> getOld(final RyaURI key) throws EventStorageException {
- try {
- return events.get(key);
- } catch (final ObjectStorageException e) {
- throw new EventStorageException(e.getMessage(), e);
- }
- }
-
- @Override
- public void create(final Event newObj) throws EventStorageException {
- try {
- events.create(newObj);
- } catch (final ObjectStorageException e) {
- throw new EventStorageException(e.getMessage(), e);
- }
- }
-
- @Override
- public void update(final Event old, final Event updated) throws EventStorageException {
- try {
- events.update(old, updated);
- } catch (final ObjectStorageException e) {
- throw new EventStorageException(e.getMessage(), e);
- }
- }
-
- public void delete(final Event event) throws EventStorageException {
- try {
- events.delete(event.getSubject());
- } catch (final ObjectStorageException e) {
- throw new EventStorageException(e.getMessage(), e);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/9e76b8d7/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/mongo/GeoTemporalMongoDBStorageStrategy.java
----------------------------------------------------------------------
diff --git a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/mongo/GeoTemporalMongoDBStorageStrategy.java b/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/mongo/GeoTemporalMongoDBStorageStrategy.java
deleted file mode 100644
index 6e8ed99..0000000
--- a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/mongo/GeoTemporalMongoDBStorageStrategy.java
+++ /dev/null
@@ -1,299 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.rya.indexing.geotemporal.mongo;
-
-import static org.apache.rya.indexing.mongodb.geo.GeoMongoDBStorageStrategy.GeoQueryType.EQUALS;
-import static org.apache.rya.indexing.mongodb.geo.GeoMongoDBStorageStrategy.GeoQueryType.INTERSECTS;
-import static org.apache.rya.indexing.mongodb.geo.GeoMongoDBStorageStrategy.GeoQueryType.WITHIN;
-import static org.apache.rya.indexing.mongodb.temporal.TemporalMongoDBStorageStrategy.INSTANT;
-import static org.apache.rya.indexing.mongodb.temporal.TemporalMongoDBStorageStrategy.INTERVAL_END;
-import static org.apache.rya.indexing.mongodb.temporal.TemporalMongoDBStorageStrategy.INTERVAL_START;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.regex.Matcher;
-
-import org.apache.log4j.Logger;
-import org.apache.rya.api.domain.RyaStatement;
-import org.apache.rya.api.resolver.RyaToRdfConversions;
-import org.apache.rya.indexing.GeoConstants;
-import org.apache.rya.indexing.IndexingExpr;
-import org.apache.rya.indexing.TemporalInstant;
-import org.apache.rya.indexing.TemporalInstantRfc3339;
-import org.apache.rya.indexing.TemporalInterval;
-import org.apache.rya.indexing.accumulo.geo.GeoParseUtils;
-import org.apache.rya.indexing.geotemporal.GeoTemporalIndexException;
-import org.apache.rya.indexing.geotemporal.GeoTemporalIndexer.GeoPolicy;
-import org.apache.rya.indexing.geotemporal.GeoTemporalIndexer.TemporalPolicy;
-import org.apache.rya.indexing.mongodb.IndexingMongoDBStorageStrategy;
-import org.apache.rya.indexing.mongodb.geo.GeoMongoDBStorageStrategy;
-import org.apache.rya.indexing.mongodb.geo.GeoMongoDBStorageStrategy.GeoQuery;
-import org.apache.rya.indexing.mongodb.temporal.TemporalMongoDBStorageStrategy;
-import org.joda.time.DateTime;
-import org.openrdf.model.Statement;
-import org.openrdf.model.URI;
-import org.openrdf.model.Value;
-import org.openrdf.query.MalformedQueryException;
-
-import com.mongodb.BasicDBObject;
-import com.mongodb.BasicDBObjectBuilder;
-import com.mongodb.DBCollection;
-import com.mongodb.DBObject;
-import com.mongodb.QueryBuilder;
-import com.vividsolutions.jts.geom.Geometry;
-import com.vividsolutions.jts.io.ParseException;
-import com.vividsolutions.jts.io.WKTReader;
-
-import jline.internal.Log;
-
-/**
- * Storage adapter for serializing Geo Temporal statements into mongo objects.
- * This includes adapting the {@link IndexingExpr}s for the GeoTemporal indexer.
- */
-public class GeoTemporalMongoDBStorageStrategy extends IndexingMongoDBStorageStrategy {
- private static final Logger LOG = Logger.getLogger(GeoTemporalMongoDBStorageStrategy.class);
- private static final String GEO_KEY = "location";
- private static final String TIME_KEY = "time";
- private final TemporalMongoDBStorageStrategy temporalStrategy;
- private final GeoMongoDBStorageStrategy geoStrategy;
-
- public GeoTemporalMongoDBStorageStrategy() {
- geoStrategy = new GeoMongoDBStorageStrategy(0.0);
- temporalStrategy = new TemporalMongoDBStorageStrategy();
- }
-
- @Override
- public void createIndices(final DBCollection coll){
- coll.createIndex(new BasicDBObject(GEO_KEY, "2dsphere"));
- coll.createIndex(TIME_KEY);
- }
-
- public DBObject getFilterQuery(final Collection<IndexingExpr> geoFilters, final Collection<IndexingExpr> temporalFilters) throws GeoTemporalIndexException {
- final QueryBuilder builder = QueryBuilder.start();
-
- if(!geoFilters.isEmpty()) {
- final DBObject[] geo = getGeoObjs(geoFilters);
- if(!temporalFilters.isEmpty()) {
- final DBObject[] temporal = getTemporalObjs(temporalFilters);
- builder.and(oneOrAnd(geo), oneOrAnd(temporal));
- return builder.get();
- } else {
- return oneOrAnd(geo);
- }
- } else if(!temporalFilters.isEmpty()) {
- final DBObject[] temporal = getTemporalObjs(temporalFilters);
- return oneOrAnd(temporal);
- } else {
- return builder.get();
- }
- }
-
- private DBObject oneOrAnd(final DBObject[] dbos) {
- if(dbos.length == 1) {
- return dbos[0];
- }
- return QueryBuilder.start()
- .and(dbos)
- .get();
- }
-
- @Override
- public DBObject serialize(final RyaStatement ryaStatement) {
- final BasicDBObjectBuilder builder = BasicDBObjectBuilder.start("_id", ryaStatement.getSubject().hashCode());
- final URI obj = ryaStatement.getObject().getDataType();
-
-
- if(obj.equals(GeoConstants.GEO_AS_WKT) || obj.equals(GeoConstants.GEO_AS_GML) ||
- obj.equals(GeoConstants.XMLSCHEMA_OGC_GML) || obj.equals(GeoConstants.XMLSCHEMA_OGC_WKT)) {
- try {
- final Statement statement = RyaToRdfConversions.convertStatement(ryaStatement);
- final Geometry geo = GeoParseUtils.getGeometry(statement);
- if (geo.getNumPoints() > 1) {
- builder.add(GEO_KEY, geoStrategy.getCorrespondingPoints(geo));
- } else {
- builder.add(GEO_KEY, geoStrategy.getDBPoint(geo));
- }
- } catch (final ParseException e) {
- LOG.error("Could not create geometry for statement " + ryaStatement, e);
- return null;
- }
- } else {
- builder.add(TIME_KEY, temporalStrategy.getTimeValue(ryaStatement.getObject().getData()));
- }
- return builder.get();
- }
-
- private DBObject[] getGeoObjs(final Collection<IndexingExpr> geoFilters) {
- final List<DBObject> objs = new ArrayList<>();
- geoFilters.forEach(filter -> {
- final GeoPolicy policy = GeoPolicy.fromURI(filter.getFunction());
- final WKTReader reader = new WKTReader();
- final String geoStr = ((Value) filter.getArguments()[0]).stringValue();
- try {
- //This method is what is used in the GeoIndexer.
- final Geometry geo = reader.read(geoStr);
- objs.add(getGeoObject(geo, policy));
- } catch (final GeoTemporalIndexException | UnsupportedOperationException | ParseException e) {
- Log.error("Unable to parse '" + geoStr + "'.", e);
- }
- });
- return objs.toArray(new DBObject[]{});
- }
-
- private DBObject[] getTemporalObjs(final Collection<IndexingExpr> temporalFilters) {
- final List<DBObject> objs = new ArrayList<>();
- temporalFilters.forEach(filter -> {
- final TemporalPolicy policy = TemporalPolicy.fromURI(filter.getFunction());
- final String timeStr = ((Value) filter.getArguments()[0]).stringValue();
- final Matcher matcher = TemporalInstantRfc3339.PATTERN.matcher(timeStr);
- if(matcher.find()) {
- final TemporalInterval interval = TemporalInstantRfc3339.parseInterval(timeStr);
- if(policy == TemporalPolicy.INSTANT_AFTER_INSTANT ||
- policy == TemporalPolicy.INSTANT_BEFORE_INSTANT ||
- policy == TemporalPolicy.INSTANT_EQUALS_INSTANT) {
- if(interval == null) {
- Log.error("Cannot perform temporal interval based queries on an instant.");
- }
- }
- objs.add(getTemporalObject(interval, policy));
- } else {
- final TemporalInstant instant = new TemporalInstantRfc3339(DateTime.parse(timeStr));
- if(policy != TemporalPolicy.INSTANT_AFTER_INSTANT &&
- policy != TemporalPolicy.INSTANT_BEFORE_INSTANT &&
- policy != TemporalPolicy.INSTANT_EQUALS_INSTANT) {
- Log.error("Cannot perform temporal instant based queries on an interval.");
- }
- objs.add(getTemporalObject(instant, policy));
- }
- });
- return objs.toArray(new DBObject[]{});
- }
-
- private DBObject getGeoObject (final Geometry geo, final GeoPolicy policy) throws GeoTemporalIndexException {
- switch(policy) {
- case CONTAINS:
- throw new UnsupportedOperationException("Contains queries are not supported in Mongo DB.");
- case CROSSES:
- throw new UnsupportedOperationException("Crosses queries are not supported in Mongo DB.");
- case DISJOINT:
- throw new UnsupportedOperationException("Disjoint queries are not supported in Mongo DB.");
- case EQUALS:
- try {
- return geoStrategy.getQuery(new GeoQuery(EQUALS, geo));
- } catch (final MalformedQueryException e) {
- throw new GeoTemporalIndexException(e.getMessage(), e);
- }
- case INTERSECTS:
- try {
- return geoStrategy.getQuery(new GeoQuery(INTERSECTS, geo));
- } catch (final MalformedQueryException e) {
- throw new GeoTemporalIndexException(e.getMessage(), e);
- }
- case OVERLAPS:
- throw new UnsupportedOperationException("Overlaps queries are not supported in Mongo DB.");
- case TOUCHES:
- throw new UnsupportedOperationException("Touches queries are not supported in Mongo DB.");
- case WITHIN:
- try {
- return geoStrategy.getQuery(new GeoQuery(WITHIN, geo));
- } catch (final MalformedQueryException e) {
- throw new GeoTemporalIndexException(e.getMessage(), e);
- }
- default:
- return new BasicDBObject();
- }
- }
-
- private DBObject getTemporalObject(final TemporalInstant instant, final TemporalPolicy policy) {
- final DBObject temporalObj;
- switch(policy) {
- case INSTANT_AFTER_INSTANT:
- temporalObj = QueryBuilder.start(INSTANT)
- .greaterThan(instant.getAsDateTime().toDate())
- .get();
- break;
- case INSTANT_BEFORE_INSTANT:
- temporalObj = QueryBuilder.start(INSTANT)
- .lessThan(instant.getAsDateTime().toDate())
- .get();
- break;
- case INSTANT_EQUALS_INSTANT:
- temporalObj = QueryBuilder.start(INSTANT)
- .is(instant.getAsDateTime().toDate())
- .get();
- break;
- default:
- temporalObj = new BasicDBObject();
- }
- return temporalObj;
- }
-
- private DBObject getTemporalObject(final TemporalInterval interval, final TemporalPolicy policy) {
- final DBObject temporalObj;
- switch(policy) {
- case INSTANT_AFTER_INTERVAL:
- temporalObj = QueryBuilder.start(INSTANT)
- .greaterThan(interval.getHasEnd().getAsDateTime().toDate())
- .get();
- break;
- case INSTANT_BEFORE_INTERVAL:
- temporalObj = QueryBuilder.start(INSTANT)
- .lessThan(interval.getHasBeginning().getAsDateTime().toDate())
- .get();
- break;
- case INSTANT_END_INTERVAL:
- temporalObj = QueryBuilder.start(INSTANT)
- .is(interval.getHasEnd().getAsDateTime().toDate())
- .get();
- break;
- case INSTANT_IN_INTERVAL:
- temporalObj = QueryBuilder.start(INSTANT)
- .greaterThan(interval.getHasBeginning().getAsDateTime().toDate())
- .lessThan(interval.getHasEnd().getAsDateTime().toDate())
- .get();
- break;
- case INSTANT_START_INTERVAL:
- temporalObj = QueryBuilder.start(INSTANT)
- .is(interval.getHasBeginning().getAsDateTime().toDate())
- .get();
- break;
- case INTERVAL_AFTER:
- temporalObj = QueryBuilder.start(INTERVAL_START)
- .greaterThan(interval.getHasEnd().getAsDateTime().toDate())
- .get();
- break;
- case INTERVAL_BEFORE:
- temporalObj = QueryBuilder.start(INTERVAL_END)
- .lessThan(interval.getHasBeginning().getAsDateTime().toDate())
- .get();
- break;
- case INTERVAL_EQUALS:
- temporalObj = QueryBuilder.start(INTERVAL_START)
- .is(interval.getHasBeginning().getAsDateTime().toDate())
- .and(INTERVAL_END)
- .is(interval.getHasEnd().getAsDateTime().toDate())
- .get();
- break;
- default:
- temporalObj = new BasicDBObject();
- }
- return temporalObj;
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/9e76b8d7/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/mongo/MongoEventStorage.java
----------------------------------------------------------------------
diff --git a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/mongo/MongoEventStorage.java b/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/mongo/MongoEventStorage.java
deleted file mode 100644
index 9c13c8b..0000000
--- a/extras/rya.geoindexing/src/main/java/org/apache/rya/indexing/geotemporal/mongo/MongoEventStorage.java
+++ /dev/null
@@ -1,195 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.rya.indexing.geotemporal.mongo;
-
-import static java.util.Objects.requireNonNull;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Optional;
-import java.util.Set;
-
-import org.apache.rya.api.domain.RyaURI;
-import org.apache.rya.indexing.IndexingExpr;
-import org.apache.rya.indexing.entity.model.TypedEntity;
-import org.apache.rya.indexing.entity.storage.mongo.DocumentConverter.DocumentConverterException;
-import org.apache.rya.indexing.entity.storage.mongo.MongoEntityStorage;
-import org.apache.rya.indexing.geotemporal.GeoTemporalIndexException;
-import org.apache.rya.indexing.geotemporal.model.Event;
-import org.apache.rya.indexing.geotemporal.storage.EventStorage;
-import org.bson.BsonDocument;
-import org.bson.BsonString;
-import org.bson.Document;
-import org.bson.conversions.Bson;
-
-import com.mongodb.BasicDBObjectBuilder;
-import com.mongodb.DBObject;
-import com.mongodb.ErrorCategory;
-import com.mongodb.MongoClient;
-import com.mongodb.MongoException;
-import com.mongodb.client.MongoCollection;
-import com.mongodb.client.MongoCursor;
-import com.mongodb.client.model.Filters;
-
-public class MongoEventStorage implements EventStorage {
-
- protected static final String COLLECTION_NAME = "geotemporal-events";
-
- private static final EventDocumentConverter EVENT_CONVERTER = new EventDocumentConverter();
-
- /**
- * A client connected to the Mongo instance that hosts the Rya instance.
- */
- protected final MongoClient mongo;
-
- /**
- * The name of the Rya instance the {@link TypedEntity}s are for.
- */
- protected final String ryaInstanceName;
-
- /*
- * Used to get the filter query objects.
- */
- private final GeoTemporalMongoDBStorageStrategy queryAdapter;
-
- /**
- * Constructs an instance of {@link MongoEntityStorage}.
- *
- * @param mongo - A client connected to the Mongo instance that hosts the Rya instance. (not null)
- * @param ryaInstanceName - The name of the Rya instance the {@link TypedEntity}s are for. (not null)
- */
- public MongoEventStorage(final MongoClient mongo, final String ryaInstanceName) {
- this.mongo = requireNonNull(mongo);
- this.ryaInstanceName = requireNonNull(ryaInstanceName);
- queryAdapter = new GeoTemporalMongoDBStorageStrategy();
- }
-
- @Override
- public void create(final Event event) throws EventStorageException {
- requireNonNull(event);
-
- try {
- mongo.getDatabase(ryaInstanceName)
- .getCollection(COLLECTION_NAME)
- .insertOne(EVENT_CONVERTER.toDocument(event));
- } catch(final MongoException e) {
- final ErrorCategory category = ErrorCategory.fromErrorCode( e.getCode() );
- if(category == ErrorCategory.DUPLICATE_KEY) {
- throw new EventAlreadyExistsException("Failed to create Event with Subject '" + event.getSubject().getData() + "'.", e);
- }
- throw new EventStorageException("Failed to create Event with Subject '" + event.getSubject().getData() + "'.", e);
- }
- }
-
- @Override
- public Optional<Event> get(final RyaURI subject) throws EventStorageException {
- requireNonNull(subject);
-
- try {
- final Document document = mongo.getDatabase(ryaInstanceName)
- .getCollection(COLLECTION_NAME)
- .find( new BsonDocument(EventDocumentConverter.SUBJECT, new BsonString(subject.getData())) )
- .first();
-
- return document == null ?
- Optional.empty() :
- Optional.of( EVENT_CONVERTER.fromDocument(document) );
-
- } catch(final MongoException | DocumentConverterException e) {
- throw new EventStorageException("Could not get the Event with Subject '" + subject.getData() + "'.", e);
- }
- }
-
- @Override
- public Collection<Event> search(final Optional<RyaURI> subject, final Optional<Collection<IndexingExpr>> geoFilters, final Optional<Collection<IndexingExpr>> temporalFilters) throws EventStorageException {
- requireNonNull(subject);
-
- try {
- final Collection<IndexingExpr> geos = (geoFilters.isPresent() ? geoFilters.get() : new ArrayList<>());
- final Collection<IndexingExpr> tempos = (temporalFilters.isPresent() ? temporalFilters.get() : new ArrayList<>());
- final DBObject filterObj = queryAdapter.getFilterQuery(geos, tempos);
-
- final BasicDBObjectBuilder builder = BasicDBObjectBuilder
- .start(filterObj.toMap());
- if(subject.isPresent()) {
- builder.append(EventDocumentConverter.SUBJECT, subject.get().getData());
- }
- final MongoCursor<Document> results = mongo.getDatabase(ryaInstanceName)
- .getCollection(COLLECTION_NAME)
- .find( BsonDocument.parse(builder.get().toString()) )
- .iterator();
-
- final List<Event> events = new ArrayList<>();
- while(results.hasNext()) {
- events.add(EVENT_CONVERTER.fromDocument(results.next()));
- }
- return events;
- } catch(final MongoException | DocumentConverterException | GeoTemporalIndexException e) {
- throw new EventStorageException("Could not get the Event.", e);
- }
- }
-
- @Override
- public void update(final Event old, final Event updated) throws StaleUpdateException, EventStorageException {
- requireNonNull(old);
- requireNonNull(updated);
-
- // The updated entity must have the same Subject as the one it is replacing.
- if(!old.getSubject().equals(updated.getSubject())) {
- throw new EventStorageException("The old Event and the updated Event must have the same Subject. " +
- "Old Subject: " + old.getSubject().getData() + ", Updated Subject: " + updated.getSubject().getData());
- }
-
- final Set<Bson> filters = new HashSet<>();
-
- // Must match the old entity's Subject.
- filters.add( makeSubjectFilter(old.getSubject()) );
-
- // Do a find and replace.
- final Bson oldEntityFilter = Filters.and(filters);
- final Document updatedDoc = EVENT_CONVERTER.toDocument(updated);
-
- final MongoCollection<Document> collection = mongo.getDatabase(ryaInstanceName).getCollection(COLLECTION_NAME);
- if(collection.findOneAndReplace(oldEntityFilter, updatedDoc) == null) {
- throw new StaleUpdateException("Could not update the Event with Subject '" + updated.getSubject().getData() + ".");
- }
- }
-
- @Override
- public boolean delete(final RyaURI subject) throws EventStorageException {
- requireNonNull(subject);
-
- try {
- final Document deleted = mongo.getDatabase(ryaInstanceName)
- .getCollection(COLLECTION_NAME)
- .findOneAndDelete( makeSubjectFilter(subject) );
-
- return deleted != null;
-
- } catch(final MongoException e) {
- throw new EventStorageException("Could not delete the Event with Subject '" + subject.getData() + "'.", e);
- }
- }
-
- private static Bson makeSubjectFilter(final RyaURI subject) {
- return Filters.eq(EventDocumentConverter.SUBJECT, subject.getData());
- }
-}