extras/indexing/src/main/java/org/apache/rya/indexing/TemporalTupleSet.java - rya - Git at Google

 /*
  * 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;

 import java.util.Map;
 import java.util.Set;

 import org.apache.hadoop.conf.Configuration;
 import org.apache.rya.indexing.external.tupleSet.ExternalTupleSet;
 import org.eclipse.rdf4j.common.iteration.CloseableIteration;
 import org.eclipse.rdf4j.model.IRI;
 import org.eclipse.rdf4j.model.Statement;
 import org.eclipse.rdf4j.model.Value;
 import org.eclipse.rdf4j.model.ValueFactory;
 import org.eclipse.rdf4j.model.impl.SimpleValueFactory;
 import org.eclipse.rdf4j.query.BindingSet;
 import org.eclipse.rdf4j.query.QueryEvaluationException;
 import org.eclipse.rdf4j.query.algebra.QueryModelVisitor;
 import org.joda.time.DateTime;

 import com.google.common.base.Joiner;
 import com.google.common.collect.Maps;

 /**
  * Indexing Node for temporal expressions to be inserted into execution plan
  * to delegate temporal portion of query to temporal index
  */
 public class TemporalTupleSet extends ExternalTupleSet {
     private static final long serialVersionUID = 1L;

     private static final ValueFactory VF = SimpleValueFactory.getInstance();

     private final Configuration conf;
     private final TemporalIndexer temporalIndexer;
     private final IndexingExpr filterInfo;

     public TemporalTupleSet(final IndexingExpr filterInfo, final TemporalIndexer temporalIndexer) {
         this.filterInfo = filterInfo;
         this.temporalIndexer = temporalIndexer;
         conf = temporalIndexer.getConf();
     }

     /**
      * {@inheritDoc}
      */
     @Override
     public Set<String> getBindingNames() {
         return filterInfo.getBindingNames();
     }

     /**
      * {@inheritDoc}
      * <p>
      * Note that we need a deep copy for everything that (during optimizations)
      * can be altered via {@link #visitChildren(QueryModelVisitor)}
      */
     @Override
     public TemporalTupleSet clone() {
         return new TemporalTupleSet(filterInfo, temporalIndexer);
     }

     @Override
     public double cardinality() {
         return 0.0; // No idea how the estimate cardinality here.
     }

     @Override
     public String getSignature() {

         return "(TemporalTuple Projection) " + "variables: " + Joiner.on(", ").join(getBindingNames()).replaceAll("\\s+", " ");
     }

     @Override
     public boolean equals(final Object other) {
         if (other == this) {
             return true;
         }
         if (!(other instanceof TemporalTupleSet)) {
             return false;
         }
         final TemporalTupleSet arg = (TemporalTupleSet) other;
         return filterInfo.equals(arg.filterInfo);
     }

     @Override
     public int hashCode() {
         int result = 17;
         result = 31*result + filterInfo.hashCode();

         return result;
     }

     /**
      * Returns an iterator over the result set associated with contained IndexingExpr.
      * <p>
      * Should be thread-safe (concurrent invocation {@link OfflineIterable} this
      * method can be expected with some query evaluators.
      */
     @Override
     public CloseableIteration<BindingSet, QueryEvaluationException> evaluate(final BindingSet bindings)
             throws QueryEvaluationException {
         final IRI funcURI = filterInfo.getFunction();
         final SearchFunction searchFunction = new TemporalSearchFunctionFactory(conf, temporalIndexer).getSearchFunction(funcURI);

         if(filterInfo.getArguments().length > 1) {
             throw new IllegalArgumentException("Index functions do not support more than two arguments.");
         }

         final String queryText = ((Value) filterInfo.getArguments()[0]).stringValue();
         return IteratorFactory.getIterator(filterInfo.getSpConstraint(), bindings, queryText, searchFunction);
     }

     //returns appropriate search function for a given URI
     //search functions used by TemporalIndexer to query Temporal Index
     public static class TemporalSearchFunctionFactory  {
         private final Map<IRI, SearchFunction> SEARCH_FUNCTION_MAP = Maps.newHashMap();
         private final TemporalIndexer temporalIndexer;

         public TemporalSearchFunctionFactory(final Configuration conf, final TemporalIndexer temporalIndexer) {
             this.temporalIndexer = temporalIndexer;
         }

         /**
          * Get a {@link TemporalSearchFunction} for a give URI.
          *
          * @param searchFunction
          * @return
          */
         public SearchFunction getSearchFunction(final IRI searchFunction) {
             SearchFunction geoFunc = null;
             try {
                 geoFunc = getSearchFunctionInternal(searchFunction);
             } catch (final QueryEvaluationException e) {
                 e.printStackTrace();
             }

             return geoFunc;
         }

         private SearchFunction getSearchFunctionInternal(final IRI searchFunction) throws QueryEvaluationException {
             final SearchFunction sf = SEARCH_FUNCTION_MAP.get(searchFunction);

             if (sf != null) {
                 return sf;
             } else {
                 throw new QueryEvaluationException("Unknown Search Function: " + searchFunction.stringValue());
             }
         }

         private final SearchFunction TEMPORAL_InstantAfterInstant = new SearchFunction() {
             @Override
             public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String searchTerms,
                     final StatementConstraints contraints) throws QueryEvaluationException {
                 final TemporalInstant queryInstant = new TemporalInstantRfc3339(DateTime.parse(searchTerms));
                 return temporalIndexer.queryInstantAfterInstant(queryInstant, contraints);
             }

             @Override
             public String toString() {
                 return "TEMPORAL_InstantAfterInstant";
             }
         };
         private final SearchFunction TEMPORAL_InstantBeforeInstant = new SearchFunction() {
             @Override
             public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String searchTerms,
                     final StatementConstraints contraints) throws QueryEvaluationException {
                 final TemporalInstant queryInstant = new TemporalInstantRfc3339(DateTime.parse(searchTerms));
                 return temporalIndexer.queryInstantBeforeInstant(queryInstant, contraints);
             }

             @Override
             public String toString() {
                 return "TEMPORAL_InstantBeforeInstant";
             }
         };

         private final SearchFunction TEMPORAL_InstantEqualsInstant = new SearchFunction() {
             @Override
             public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String searchTerms,
                     final StatementConstraints contraints) throws QueryEvaluationException {
                 final TemporalInstant queryInstant = new TemporalInstantRfc3339(DateTime.parse(searchTerms));
                 return temporalIndexer.queryInstantEqualsInstant(queryInstant, contraints);
             }

             @Override
             public String toString() {
                 return "TEMPORAL_InstantEqualsInstant";
             }
         };

         private final SearchFunction TEMPORAL_InstantAfterInterval = new SearchFunction() {
             @Override
             public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String searchTerms,
                     final StatementConstraints contraints) throws QueryEvaluationException {
                 final TemporalInterval queryInterval = TemporalInstantRfc3339.parseInterval(searchTerms);
                 return temporalIndexer.queryInstantAfterInterval(queryInterval, contraints);
             }

             @Override
             public String toString() {
                 return "TEMPORAL_InstantAfterInterval";
             }
         };

         private final SearchFunction TEMPORAL_InstantBeforeInterval = new SearchFunction() {
             @Override
             public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String searchTerms,
                     final StatementConstraints contraints) throws QueryEvaluationException {
                 final TemporalInterval queryInterval = TemporalInstantRfc3339.parseInterval(searchTerms);
                 return temporalIndexer.queryInstantBeforeInterval(queryInterval, contraints);
             }

             @Override
             public String toString() {
                 return "TEMPORAL_InstantBeforeInterval";
             }
         };

         private final SearchFunction TEMPORAL_InstantInsideInterval = new SearchFunction() {
             @Override
             public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String searchTerms,
                     final StatementConstraints contraints) throws QueryEvaluationException {
                 final TemporalInterval queryInterval = TemporalInstantRfc3339.parseInterval(searchTerms);
                 return temporalIndexer.queryInstantInsideInterval(queryInterval, contraints);
             }

             @Override
             public String toString() {
                 return "TEMPORAL_InstantInsideInterval";
             }
         };

         private final SearchFunction TEMPORAL_InstantHasBeginningInterval = new SearchFunction() {
             @Override
             public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String searchTerms,
                     final StatementConstraints contraints) throws QueryEvaluationException {
                 final TemporalInterval queryInterval = TemporalInstantRfc3339.parseInterval(searchTerms);
                 return temporalIndexer.queryInstantHasBeginningInterval(queryInterval, contraints);
             }

             @Override
             public String toString() {
                 return "TEMPORAL_InstantHasBeginningInterval";
             }
         };

         private final SearchFunction TEMPORAL_InstantHasEndInterval = new SearchFunction() {
             @Override
             public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String searchTerms,
                     final StatementConstraints contraints) throws QueryEvaluationException {
                 final TemporalInterval queryInterval = TemporalInstantRfc3339.parseInterval(searchTerms);
                 return temporalIndexer.queryInstantHasEndInterval(queryInterval, contraints);
             }

             @Override
             public String toString() {
                 return "TEMPORAL_InstantHasEndInterval";
             }
         };

         {
             final String TEMPORAL_NS = "tag:rya-rdf.org,2015:temporal#";

             SEARCH_FUNCTION_MAP.put(VF.createIRI(TEMPORAL_NS+"after"), TEMPORAL_InstantAfterInstant);
             SEARCH_FUNCTION_MAP.put(VF.createIRI(TEMPORAL_NS+"before"), TEMPORAL_InstantBeforeInstant);
             SEARCH_FUNCTION_MAP.put(VF.createIRI(TEMPORAL_NS+"equals"), TEMPORAL_InstantEqualsInstant);

             SEARCH_FUNCTION_MAP.put(VF.createIRI(TEMPORAL_NS+"beforeInterval"), TEMPORAL_InstantBeforeInterval);
             SEARCH_FUNCTION_MAP.put(VF.createIRI(TEMPORAL_NS+"afterInterval"), TEMPORAL_InstantAfterInterval);
             SEARCH_FUNCTION_MAP.put(VF.createIRI(TEMPORAL_NS+"insideInterval"), TEMPORAL_InstantInsideInterval);
             SEARCH_FUNCTION_MAP.put(VF.createIRI(TEMPORAL_NS+"hasBeginningInterval"),
                     TEMPORAL_InstantHasBeginningInterval);
             SEARCH_FUNCTION_MAP.put(VF.createIRI(TEMPORAL_NS+"hasEndInterval"), TEMPORAL_InstantHasEndInterval);
         }
     }
 }
	/*
	* 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;

	import java.util.Map;
	import java.util.Set;

	import org.apache.hadoop.conf.Configuration;
	import org.apache.rya.indexing.external.tupleSet.ExternalTupleSet;
	import org.eclipse.rdf4j.common.iteration.CloseableIteration;
	import org.eclipse.rdf4j.model.IRI;
	import org.eclipse.rdf4j.model.Statement;
	import org.eclipse.rdf4j.model.Value;
	import org.eclipse.rdf4j.model.ValueFactory;
	import org.eclipse.rdf4j.model.impl.SimpleValueFactory;
	import org.eclipse.rdf4j.query.BindingSet;
	import org.eclipse.rdf4j.query.QueryEvaluationException;
	import org.eclipse.rdf4j.query.algebra.QueryModelVisitor;
	import org.joda.time.DateTime;

	import com.google.common.base.Joiner;
	import com.google.common.collect.Maps;

	/**
	* Indexing Node for temporal expressions to be inserted into execution plan
	* to delegate temporal portion of query to temporal index
	*/
	public class TemporalTupleSet extends ExternalTupleSet {
	private static final long serialVersionUID = 1L;

	private static final ValueFactory VF = SimpleValueFactory.getInstance();

	private final Configuration conf;
	private final TemporalIndexer temporalIndexer;
	private final IndexingExpr filterInfo;

	public TemporalTupleSet(final IndexingExpr filterInfo, final TemporalIndexer temporalIndexer) {
	this.filterInfo = filterInfo;
	this.temporalIndexer = temporalIndexer;
	conf = temporalIndexer.getConf();
	}

	/**
	* {@inheritDoc}
	*/
	@Override
	public Set<String> getBindingNames() {
	return filterInfo.getBindingNames();
	}

	/**
	* {@inheritDoc}
	* <p>
	* Note that we need a deep copy for everything that (during optimizations)
	* can be altered via {@link #visitChildren(QueryModelVisitor)}
	*/
	@Override
	public TemporalTupleSet clone() {
	return new TemporalTupleSet(filterInfo, temporalIndexer);
	}

	@Override
	public double cardinality() {
	return 0.0; // No idea how the estimate cardinality here.
	}

	@Override
	public String getSignature() {

	return "(TemporalTuple Projection) " + "variables: " + Joiner.on(", ").join(getBindingNames()).replaceAll("\\s+", " ");
	}

	@Override
	public boolean equals(final Object other) {
	if (other == this) {
	return true;
	}
	if (!(other instanceof TemporalTupleSet)) {
	return false;
	}
	final TemporalTupleSet arg = (TemporalTupleSet) other;
	return filterInfo.equals(arg.filterInfo);
	}

	@Override
	public int hashCode() {
	int result = 17;
	result = 31*result + filterInfo.hashCode();

	return result;
	}

	/**
	* Returns an iterator over the result set associated with contained IndexingExpr.
	* <p>
	* Should be thread-safe (concurrent invocation {@link OfflineIterable} this
	* method can be expected with some query evaluators.
	*/
	@Override
	public CloseableIteration<BindingSet, QueryEvaluationException> evaluate(final BindingSet bindings)
	throws QueryEvaluationException {
	final IRI funcURI = filterInfo.getFunction();
	final SearchFunction searchFunction = new TemporalSearchFunctionFactory(conf, temporalIndexer).getSearchFunction(funcURI);

	if(filterInfo.getArguments().length > 1) {
	throw new IllegalArgumentException("Index functions do not support more than two arguments.");
	}

	final String queryText = ((Value) filterInfo.getArguments()[0]).stringValue();
	return IteratorFactory.getIterator(filterInfo.getSpConstraint(), bindings, queryText, searchFunction);
	}

	//returns appropriate search function for a given URI
	//search functions used by TemporalIndexer to query Temporal Index
	public static class TemporalSearchFunctionFactory {
	private final Map<IRI, SearchFunction> SEARCH_FUNCTION_MAP = Maps.newHashMap();
	private final TemporalIndexer temporalIndexer;

	public TemporalSearchFunctionFactory(final Configuration conf, final TemporalIndexer temporalIndexer) {
	this.temporalIndexer = temporalIndexer;
	}

	/**
	* Get a {@link TemporalSearchFunction} for a give URI.
	*
	* @param searchFunction
	* @return
	*/
	public SearchFunction getSearchFunction(final IRI searchFunction) {
	SearchFunction geoFunc = null;
	try {
	geoFunc = getSearchFunctionInternal(searchFunction);
	} catch (final QueryEvaluationException e) {
	e.printStackTrace();
	}

	return geoFunc;
	}

	private SearchFunction getSearchFunctionInternal(final IRI searchFunction) throws QueryEvaluationException {
	final SearchFunction sf = SEARCH_FUNCTION_MAP.get(searchFunction);

	if (sf != null) {
	return sf;
	} else {
	throw new QueryEvaluationException("Unknown Search Function: " + searchFunction.stringValue());
	}
	}

	private final SearchFunction TEMPORAL_InstantAfterInstant = new SearchFunction() {
	@Override
	public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String searchTerms,
	final StatementConstraints contraints) throws QueryEvaluationException {
	final TemporalInstant queryInstant = new TemporalInstantRfc3339(DateTime.parse(searchTerms));
	return temporalIndexer.queryInstantAfterInstant(queryInstant, contraints);
	}

	@Override
	public String toString() {
	return "TEMPORAL_InstantAfterInstant";
	}
	};
	private final SearchFunction TEMPORAL_InstantBeforeInstant = new SearchFunction() {
	@Override
	public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String searchTerms,
	final StatementConstraints contraints) throws QueryEvaluationException {
	final TemporalInstant queryInstant = new TemporalInstantRfc3339(DateTime.parse(searchTerms));
	return temporalIndexer.queryInstantBeforeInstant(queryInstant, contraints);
	}

	@Override
	public String toString() {
	return "TEMPORAL_InstantBeforeInstant";
	}
	};

	private final SearchFunction TEMPORAL_InstantEqualsInstant = new SearchFunction() {
	@Override
	public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String searchTerms,
	final StatementConstraints contraints) throws QueryEvaluationException {
	final TemporalInstant queryInstant = new TemporalInstantRfc3339(DateTime.parse(searchTerms));
	return temporalIndexer.queryInstantEqualsInstant(queryInstant, contraints);
	}

	@Override
	public String toString() {
	return "TEMPORAL_InstantEqualsInstant";
	}
	};

	private final SearchFunction TEMPORAL_InstantAfterInterval = new SearchFunction() {
	@Override
	public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String searchTerms,
	final StatementConstraints contraints) throws QueryEvaluationException {
	final TemporalInterval queryInterval = TemporalInstantRfc3339.parseInterval(searchTerms);
	return temporalIndexer.queryInstantAfterInterval(queryInterval, contraints);
	}

	@Override
	public String toString() {
	return "TEMPORAL_InstantAfterInterval";
	}
	};

	private final SearchFunction TEMPORAL_InstantBeforeInterval = new SearchFunction() {
	@Override
	public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String searchTerms,
	final StatementConstraints contraints) throws QueryEvaluationException {
	final TemporalInterval queryInterval = TemporalInstantRfc3339.parseInterval(searchTerms);
	return temporalIndexer.queryInstantBeforeInterval(queryInterval, contraints);
	}

	@Override
	public String toString() {
	return "TEMPORAL_InstantBeforeInterval";
	}
	};

	private final SearchFunction TEMPORAL_InstantInsideInterval = new SearchFunction() {
	@Override
	public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String searchTerms,
	final StatementConstraints contraints) throws QueryEvaluationException {
	final TemporalInterval queryInterval = TemporalInstantRfc3339.parseInterval(searchTerms);
	return temporalIndexer.queryInstantInsideInterval(queryInterval, contraints);
	}

	@Override
	public String toString() {
	return "TEMPORAL_InstantInsideInterval";
	}
	};

	private final SearchFunction TEMPORAL_InstantHasBeginningInterval = new SearchFunction() {
	@Override
	public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String searchTerms,
	final StatementConstraints contraints) throws QueryEvaluationException {
	final TemporalInterval queryInterval = TemporalInstantRfc3339.parseInterval(searchTerms);
	return temporalIndexer.queryInstantHasBeginningInterval(queryInterval, contraints);
	}

	@Override
	public String toString() {
	return "TEMPORAL_InstantHasBeginningInterval";
	}
	};

	private final SearchFunction TEMPORAL_InstantHasEndInterval = new SearchFunction() {
	@Override
	public CloseableIteration<Statement, QueryEvaluationException> performSearch(final String searchTerms,
	final StatementConstraints contraints) throws QueryEvaluationException {
	final TemporalInterval queryInterval = TemporalInstantRfc3339.parseInterval(searchTerms);
	return temporalIndexer.queryInstantHasEndInterval(queryInterval, contraints);
	}

	@Override
	public String toString() {
	return "TEMPORAL_InstantHasEndInterval";
	}
	};

	{
	final String TEMPORAL_NS = "tag:rya-rdf.org,2015:temporal#";

	SEARCH_FUNCTION_MAP.put(VF.createIRI(TEMPORAL_NS+"after"), TEMPORAL_InstantAfterInstant);
	SEARCH_FUNCTION_MAP.put(VF.createIRI(TEMPORAL_NS+"before"), TEMPORAL_InstantBeforeInstant);
	SEARCH_FUNCTION_MAP.put(VF.createIRI(TEMPORAL_NS+"equals"), TEMPORAL_InstantEqualsInstant);

	SEARCH_FUNCTION_MAP.put(VF.createIRI(TEMPORAL_NS+"beforeInterval"), TEMPORAL_InstantBeforeInterval);
	SEARCH_FUNCTION_MAP.put(VF.createIRI(TEMPORAL_NS+"afterInterval"), TEMPORAL_InstantAfterInterval);
	SEARCH_FUNCTION_MAP.put(VF.createIRI(TEMPORAL_NS+"insideInterval"), TEMPORAL_InstantInsideInterval);
	SEARCH_FUNCTION_MAP.put(VF.createIRI(TEMPORAL_NS+"hasBeginningInterval"),
	TEMPORAL_InstantHasBeginningInterval);
	SEARCH_FUNCTION_MAP.put(VF.createIRI(TEMPORAL_NS+"hasEndInterval"), TEMPORAL_InstantHasEndInterval);
	}
	}
	}