DATAREDIS-425 - Make sure index clean up removes values from all potential indexes.

Author: christophstrobl

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

@@ -34,6 +34,7 @@
 import org.springframework.data.redis.core.convert.RedisData;
 import org.springframework.data.redis.core.convert.ReferenceResolverImpl;
 import org.springframework.data.redis.core.index.IndexConfiguration;
+import org.springframework.data.redis.util.ByteUtils;
 import org.springframework.data.util.CloseableIterator;
 
 /**
@@ -88,6 +89,8 @@ public Object put(final Serializable id, final Object item, final Serializable k
 		final RedisData rdo = new RedisData();
 		converter.write(item, rdo);
 
+		final byte[] indexPostFixPattern = converter.toBytes(":*");
+
 		redisOps.execute(new RedisCallback<Object>() {
 
 			@Override
@@ -96,7 +99,18 @@ public Object doInRedis(RedisConnection connection) throws DataAccessException {
 				connection.hMSet(rdo.getKey(), rdo.getData());
 				connection.sAdd(rdo.getKeyspace(), rdo.getId());
 
+				// remove id from potential indexes since those might be invalid with the new data
+				for (byte[] potentialIndex : rdo.getIndexPaths()) {
+
+					Set<byte[]> existingKeys = connection.keys(ByteUtils.concat(potentialIndex, indexPostFixPattern));
+
+					for (byte[] existingKey : existingKeys) {
+						connection.sRem(existingKey, rdo.getId());
+					}
+				}
+
 				if (!rdo.getSimpleIndexKeys().isEmpty()) {
+
 					for (byte[] index : rdo.getSimpleIndexKeys()) {
 						connection.sAdd(index, rdo.getId());
 					}

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

@@ -343,23 +343,31 @@ public void doWithPersistentProperty(KeyValuePersistentProperty persistentProper
 							.getTypeInformation().getActualType(), sink);
 				} else {
 
-					if (indexConfiguration.hasIndexFor(entity.getKeySpace(), persistentProperty.getName())) {
+					Object propertyValue = accessor.getProperty(persistentProperty);
 
-						// TODO: check all index types and add accordingly
-						sink.addSimpleIndexKey(toBytes(persistentProperty.getName() + ":"
-								+ accessor.getProperty(persistentProperty)));
+					if (indexConfiguration.hasIndexFor(entity.getKeySpace(), propertyStringPath)) {
+
+						if (propertyValue != null) {
+							sink.addSimpleIndexKey(toBytes(propertyStringPath + ":" + propertyValue));
+						}
+
+						sink.addIndexPath(toBytes(propertyStringPath));
 					}
 
 					else if (persistentProperty.isAnnotationPresent(Indexed.class)) {
 
 						// TOOD: read index type from annotation
-						indexConfiguration.addIndexDefinition(new RedisIndexDefinition(entity.getKeySpace(), persistentProperty
-								.getName(), IndexType.SIMPLE));
+						indexConfiguration.addIndexDefinition(new RedisIndexDefinition(entity.getKeySpace(), propertyStringPath,
+								IndexType.SIMPLE));
 
-						sink.addSimpleIndexKey(toBytes(persistentProperty.getName() + ":"
-								+ accessor.getProperty(persistentProperty)));
+						if (propertyValue != null) {
+							sink.addSimpleIndexKey(toBytes(propertyStringPath + ":" + propertyValue));
+						}
+
+						sink.addIndexPath(toBytes(propertyStringPath));
 					}
-					sink.addDataEntry(toBytes(propertyStringPath), toBytes(accessor.getProperty(persistentProperty)));
+
+					sink.addDataEntry(toBytes(propertyStringPath), toBytes(propertyValue));
 				}
 			}
 		});

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

@@ -42,11 +42,13 @@ public class RedisData {
 
 	private Map<ByteArrayWrapper, byte[]> data;
 	private Set<ByteArrayWrapper> simpleIndexKeys;
+	private Set<ByteArrayWrapper> indexPaths;
 
 	public RedisData() {
 
 		this.data = new LinkedHashMap<ByteArrayWrapper, byte[]>();
 		this.simpleIndexKeys = new HashSet<ByteArrayWrapper>();
+		this.indexPaths = new HashSet<ByteArrayWrapper>();
 	}
 
 	public RedisData(Map<byte[], byte[]> raw) {
@@ -137,6 +139,10 @@ public void addSimpleIndexKey(byte[] bytes) {
 		this.simpleIndexKeys.add(new ByteArrayWrapper(bytes));
 	}
 
+	public void addIndexPath(byte[] path) {
+		this.indexPaths.add(new ByteArrayWrapper(path));
+	}
+
 	public Set<byte[]> getSimpleIndexKeys() {
 
 		Set<byte[]> target = new HashSet<byte[]>();
@@ -146,6 +152,15 @@ public Set<byte[]> getSimpleIndexKeys() {
 		return target;
 	}
 
+	public Set<byte[]> getIndexPaths() {
+
+		Set<byte[]> target = new HashSet<byte[]>();
+		for (ByteArrayWrapper wrapper : this.indexPaths) {
+			target.add(ByteUtils.concatAll(keyspace, PATH_SEPERATOR, wrapper.getArray()));
+		}
+		return target;
+	}
+
 	/**
 	 * @return
 	 */