DATAREDIS-533 - Add support for geo indexes.

We now allow usage of @GeoIndexed to mark GeoLocation or Point properties as candidates for secondary index creation. Non null values will be included in GEOADD command as follows:

    GEOADD keyspace:property-path point.x point.y entity-id

@GeoIndexed can be used on top level as well as on nested properties.

class Person {

	@id String id;
	String firstname, lastname;
	Address hometown;
}

class Address {
	String city, street, housenumber;
	@GeoIndexed Point location;
}

The above allows us to derive geospatial queries from a given method using NEAR and WITHIN keywords like:

interface PersonRepository extends CrudRepository<Person, String> {

	List<Person> findByAddressLocationNear(Point point, Distance distance);

	List<Person> findByAddressLocationWithin(Circle circle);
}

Partial updates on the Point itself also trigger an index refresh operation. So it is possible to alter existing entities via:

template.save(new PartialUpdate<Person>("1", Person.class).set("address.location", new Point(17, 18));

Original pull request: spring-projects#215.

Author: christophstrobl

src/main/asciidoc/reference/redis-repositories.adoc

@@ -310,6 +310,8 @@ public class ApplicationConfig {
 == Secondary Indexes
 http://redis.io/topics/indexes[Secondary indexes] are used to enable lookup operations based on native Redis structures. Values are written to the according indexes on every save and are removed when objects are deleted or <<redis.repositories.expirations,expire>>.
 
+=== Simple Property Index
+
 Given the sample `Person` entity we can create an index for _firstname_ by annotating the property with `@Indexed`.
 
 .Annotation driven indexing
@@ -419,6 +421,51 @@ public class ApplicationConfig {
 ----
 ====
 
+=== Geospatial Index
+
+Assume the `Address` type contains a property `location` of type `Point` that holds the geo coordinates of the particular address. By annotating the property with `@GeoIndexed` those values will be added using  Redis `GEO` commands.
+
+====
+[source,java]
+----
+@RedisHash("persons")
+public class Person {
+
+  Address address;
+
+  // ... other properties omitted
+}
+
+public class Address {
+
+  @GeoIndexed Point location;
+
+  // ... other properties omitted
+}
+
+public interface PersonRepository extends CrudRepository<Person, String> {
+
+  List<Person> findByAddressLocationNear(Point point, Distance distance);     <1>
+  List<Person> findByAddressLocationWithin(Circle circle);                    <2>
+}
+
+Person rand = new Person("rand", "al'thor");
+rand.setAddress(new Address(new Point(13.361389D, 38.115556D)));
+
+repository.save(rand);                                                        <3>
+
+repository.findByAddressLocationNear(new Point(15D, 37D), new Distance(200)); <4>
+----
+<1> finder declaration on nested property using Point and Distance.
+<2> finder declaration on nested property using Circle to search within.
+<3> `GEOADD persons:address:location 13.361389 38.115556 e2c7dcee-b8cd-4424-883e-736ce564363e`
+<4> `GEORADIUS persons:address:location 15.0 37.0 200.0 km`
+====
+
+In the above example the lon/lat values are stored using `GEOADD` using the objects `id` as the members name. The finder methods allow usage of `Circle` or `Point, Distance` combinations for querying those values.
+
+NOTE: It is **not** possible to combine `near`/`within` with other criteria.
+
 
 [[redis.repositories.expirations]]
 == Time To Live

src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java

@@ -2431,7 +2431,7 @@ public Long geoAdd(String key, Map<String, Point> memberCoordinateMap) {
 
 		Map<byte[], Point> byteMap = new HashMap<byte[], Point>();
 		for (Entry<String, Point> entry : memberCoordinateMap.entrySet()) {
-			byteMap.put(serialize(entry.getKey()), memberCoordinateMap.get(entry.getValue()));
+			byteMap.put(serialize(entry.getKey()), entry.getValue());
 		}
 
 		return geoAdd(serialize(key), byteMap);

src/main/java/org/springframework/data/redis/core/IndexWriter.java

@@ -18,7 +18,9 @@
 import java.util.Set;
 
 import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.data.redis.connection.DataType;
 import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.core.convert.GeoIndexedPropertyValue;
 import org.springframework.data.redis.core.convert.IndexedData;
 import org.springframework.data.redis.core.convert.RedisConverter;
 import org.springframework.data.redis.core.convert.RemoveIndexedData;
@@ -126,7 +128,13 @@ public void removeKeyFromIndexes(String keyspace, Object key) {
 		byte[] indexHelperKey = ByteUtils.concatAll(toBytes(keyspace + ":"), binKey, toBytes(":idx"));
 
 		for (byte[] indexKey : connection.sMembers(indexHelperKey)) {
-			connection.sRem(indexKey, binKey);
+
+			DataType type = connection.type(indexKey);
+			if (DataType.ZSET.equals(type)) {
+				connection.zRem(indexKey, binKey);
+			} else {
+				connection.sRem(indexKey, binKey);
+			}
 		}
 
 		connection.del(indexHelperKey);
@@ -166,7 +174,12 @@ protected void removeKeyFromExistingIndexes(byte[] key, IndexedData indexedData)
 
 		if (!CollectionUtils.isEmpty(existingKeys)) {
 			for (byte[] existingKey : existingKeys) {
-				connection.sRem(existingKey, key);
+
+				if (indexedData instanceof GeoIndexedPropertyValue) {
+					connection.geoRemove(existingKey, key);
+				} else {
+					connection.sRem(existingKey, key);
+				}
 			}
 		}
 	}
@@ -207,7 +220,23 @@ else if (indexedData instanceof SimpleIndexedPropertyValue) {
 
 			// keep track of indexes used for the object
 			connection.sAdd(ByteUtils.concatAll(toBytes(indexedData.getKeyspace() + ":"), key, toBytes(":idx")), indexKey);
-		} else {
+		} else if (indexedData instanceof GeoIndexedPropertyValue) {
+
+			GeoIndexedPropertyValue geoIndexedData = ((GeoIndexedPropertyValue) indexedData);
+
+			Object value = geoIndexedData.getValue();
+			if (value == null) {
+				return;
+			}
+
+			byte[] indexKey = toBytes(indexedData.getKeyspace() + ":" + indexedData.getIndexName());
+			connection.geoAdd(indexKey, geoIndexedData.getPoint(), key);
+
+			// keep track of indexes used for the object
+			connection.sAdd(ByteUtils.concatAll(toBytes(indexedData.getKeyspace() + ":"), key, toBytes(":idx")), indexKey);
+		}
+
+		else {
 			throw new IllegalArgumentException(
 					String.format("Cannot write index data for unknown index type %s", indexedData.getClass()));
 		}

src/main/java/org/springframework/data/redis/core/RedisKeyValueAdapter.java

@@ -42,12 +42,14 @@
 import org.springframework.data.keyvalue.core.KeyValueAdapter;
 import org.springframework.data.keyvalue.core.mapping.KeyValuePersistentProperty;
 import org.springframework.data.mapping.PersistentProperty;
+import org.springframework.data.redis.connection.DataType;
 import org.springframework.data.redis.connection.Message;
 import org.springframework.data.redis.connection.MessageListener;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.core.PartialUpdate.PropertyUpdate;
 import org.springframework.data.redis.core.PartialUpdate.UpdateCommand;
 import org.springframework.data.redis.core.convert.CustomConversions;
+import org.springframework.data.redis.core.convert.GeoIndexedPropertyValue;
 import org.springframework.data.redis.core.convert.KeyspaceConfiguration;
 import org.springframework.data.redis.core.convert.MappingRedisConverter;
 import org.springframework.data.redis.core.convert.PathIndexResolver;
@@ -461,8 +463,13 @@ public Void doInRedis(RedisConnection connection) throws DataAccessException {
 							redisUpdateObject.fieldsToRemove.toArray(new byte[redisUpdateObject.fieldsToRemove.size()][]));
 				}
 
-				for (byte[] index : redisUpdateObject.indexesToUpdate) {
-					connection.sRem(index, toBytes(redisUpdateObject.targetId));
+				for (RedisUpdateObject.Index index : redisUpdateObject.indexesToUpdate) {
+
+					if (ObjectUtils.nullSafeEquals(DataType.ZSET, index.type)) {
+						connection.zRem(index.key, toBytes(redisUpdateObject.targetId));
+					} else {
+						connection.sRem(index.key, toBytes(redisUpdateObject.targetId));
+					}
 				}
 
 				if (!rdo.getBucket().isEmpty()) {
@@ -509,7 +516,8 @@ private RedisUpdateObject fetchDeletePathsFromHashAndUpdateIndex(RedisUpdateObje
 					? ByteUtils.concatAll(toBytes(redisUpdateObject.keyspace), toBytes((":" + path)), toBytes(":"), value) : null;
 
 			if (connection.exists(existingValueIndexKey)) {
-				redisUpdateObject.addIndexToUpdate(existingValueIndexKey);
+
+				redisUpdateObject.addIndexToUpdate(new RedisUpdateObject.Index(existingValueIndexKey, DataType.SET));
 			}
 			return redisUpdateObject;
 		}
@@ -530,12 +538,22 @@ private RedisUpdateObject fetchDeletePathsFromHashAndUpdateIndex(RedisUpdateObje
 							: null;
 
 					if (connection.exists(existingValueIndexKey)) {
-						redisUpdateObject.addIndexToUpdate(existingValueIndexKey);
+						redisUpdateObject.addIndexToUpdate(new RedisUpdateObject.Index(existingValueIndexKey, DataType.SET));
 					}
 				}
 			}
 		}
 
+		String pathToUse = GeoIndexedPropertyValue.geoIndexName(path);
+		if (connection.zRank(ByteUtils.concatAll(toBytes(redisUpdateObject.keyspace), toBytes(":"), toBytes(pathToUse)),
+				toBytes(redisUpdateObject.targetId)) != null) {
+
+			redisUpdateObject
+					.addIndexToUpdate(new org.springframework.data.redis.core.RedisKeyValueAdapter.RedisUpdateObject.Index(
+							ByteUtils.concatAll(toBytes(redisUpdateObject.keyspace), toBytes(":"), toBytes(pathToUse)),
+							DataType.ZSET));
+		}
+
 		return redisUpdateObject;
 	}
 
@@ -855,7 +873,7 @@ private static class RedisUpdateObject {
 		private final byte[] targetKey;
 
 		private Set<byte[]> fieldsToRemove = new LinkedHashSet<byte[]>();
-		private Set<byte[]> indexesToUpdate = new LinkedHashSet<byte[]>();
+		private Set<Index> indexesToUpdate = new LinkedHashSet<Index>();
 
 		RedisUpdateObject(byte[] targetKey, String keyspace, Object targetId) {
 
@@ -868,8 +886,19 @@ void addFieldToRemove(byte[] field) {
 			fieldsToRemove.add(field);
 		}
 
-		void addIndexToUpdate(byte[] indexName) {
-			indexesToUpdate.add(indexName);
+		void addIndexToUpdate(Index index) {
+			indexesToUpdate.add(index);
+		}
+
+		static class Index {
+			final DataType type;
+			final byte[] key;
+
+			public Index(byte[] key, DataType type) {
+				this.key = key;
+				this.type = type;
+			}
+
 		}
 	}
 }

src/main/java/org/springframework/data/redis/core/RedisQueryEngine.java

@@ -25,13 +25,19 @@
 import java.util.Map;
 
 import org.springframework.dao.DataAccessException;
+import org.springframework.data.geo.Circle;
+import org.springframework.data.geo.GeoResult;
+import org.springframework.data.geo.GeoResults;
 import org.springframework.data.keyvalue.core.CriteriaAccessor;
 import org.springframework.data.keyvalue.core.QueryEngine;
 import org.springframework.data.keyvalue.core.SortAccessor;
 import org.springframework.data.keyvalue.core.query.KeyValueQuery;
 import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.RedisGeoCommands.GeoLocation;
+import org.springframework.data.redis.core.convert.GeoIndexedPropertyValue;
 import org.springframework.data.redis.core.convert.RedisData;
 import org.springframework.data.redis.repository.query.RedisOperationChain;
+import org.springframework.data.redis.repository.query.RedisOperationChain.NearPath;
 import org.springframework.data.redis.repository.query.RedisOperationChain.PathAndValue;
 import org.springframework.data.redis.util.ByteUtils;
 import org.springframework.util.CollectionUtils;
@@ -74,7 +80,8 @@ public <T> Collection<T> execute(final RedisOperationChain criteria, final Compa
 			final int rows, final Serializable keyspace, Class<T> type) {
 
 		if (criteria == null
-				|| (CollectionUtils.isEmpty(criteria.getOrSismember()) && CollectionUtils.isEmpty(criteria.getSismember()))) {
+				|| (CollectionUtils.isEmpty(criteria.getOrSismember()) && CollectionUtils.isEmpty(criteria.getSismember()))
+						&& criteria.getNear() == null) {
 			return (Collection<T>) getAdapter().getAllOf(keyspace, offset, rows);
 		}
 
@@ -99,6 +106,15 @@ public Map<byte[], Map<byte[], byte[]>> doInRedis(RedisConnection connection) th
 					allKeys.addAll(connection.sUnion(keys(keyspace + ":", criteria.getOrSismember())));
 				}
 
+				if (criteria.getNear() != null) {
+
+					GeoResults<GeoLocation<byte[]>> x = connection.geoRadius(geoKey(keyspace + ":", criteria.getNear()),
+							new Circle(criteria.getNear().getPoint(), criteria.getNear().getDistance()));
+					for (GeoResult<GeoLocation<byte[]>> y : x) {
+						allKeys.add(y.getContent().getName());
+					}
+				}
+
 				byte[] keyspaceBin = getAdapter().getConverter().getConversionService().convert(keyspace + ":", byte[].class);
 
 				final Map<byte[], Map<byte[], byte[]>> rawData = new LinkedHashMap<byte[], Map<byte[], byte[]>>();
@@ -195,6 +211,13 @@ private byte[][] keys(String prefix, Collection<PathAndValue> source) {
 		return keys;
 	}
 
+	private byte[] geoKey(String prefix, NearPath source) {
+
+		String path = GeoIndexedPropertyValue.geoIndexName(source.getPath());
+		return getAdapter().getConverter().getConversionService().convert(prefix + path, byte[].class);
+
+	}
+
 	/**
 	 * @author Christoph Strobl
 	 * @since 1.7

src/main/java/org/springframework/data/redis/core/convert/GeoIndexedPropertyValue.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed 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.springframework.data.redis.core.convert;
+
+import org.springframework.data.geo.Point;
+
+import lombok.Data;
+
+/**
+ * {@link IndexedData} implementation indicating storage of data within a Redis GEO structure.
+ *
+ * @author Christoph Strobl
+ * @since 1.8
+ */
+@Data
+public class GeoIndexedPropertyValue implements IndexedData {
+
+	private final String keyspace;
+	private final String indexName;
+	private final Point value;
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.convert.IndexedData#getIndexName()
+	 */
+	@Override
+	public String getIndexName() {
+		return GeoIndexedPropertyValue.geoIndexName(indexName);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.convert.IndexedData#getKeyspace()
+	 */
+	@Override
+	public String getKeyspace() {
+		return keyspace;
+	}
+
+	public Point getPoint() {
+		return value;
+	}
+
+	public static String geoIndexName(String path) {
+
+		int index = path.lastIndexOf('.');
+		if (index == -1) {
+			return path;
+		}
+		StringBuilder sb = new StringBuilder(path);
+		sb.setCharAt(index, ':');
+		return sb.toString();
+	}
+}