From 76229c10a44f76ec84462e77466d40675da2d5e9 Mon Sep 17 00:00:00 2001
From: Christoph Strobl <cstrobl@pivotal.io>
Date: Wed, 16 Mar 2016 10:25:33 +0100
Subject: [PATCH] DATAREDIS-471 - Add support for partial updates.

We now allow partial update of domain types via PartialUpdate. The according expiration times and secondary index structures are updated accordingly.

In some cases it is not necessary to load and rewrite the entire entity just to set a new value within it. A session timestamp for last active time might be such a scenario where you just want to alter one property.
`PartialUpdate` allows to define `set`, `delete` actions on existing objects while taking care of updating potential expiration times of the entity itself as well as index structures.

.Sample Partial Update
====
[source,java]
----
PartialUpdate<Person> update = new PartialUpdate<Person>("e2c7dcee", Person.class)
  .set("firstname", "mat")                                                           <1>
  .set("address.city", "emond's field")                                              <2>
  .del("age");                                                                       <3>

template.update(update);

update = new PartialUpdate<Person>("e2c7dcee", Person.class)
  .set("address", new Address("caemlyn", "andor"))                                   <4>
  .set("attributes", singletonMap("eye-color", "grey"));                             <5>

template.update(update);

update = new PartialUpdate<Person>("e2c7dcee", Person.class)
  .refreshTtl(true);                                                                 <6>
  .set("expiration", 1000);

template.update(update);
----
<1> Set the simple property _firstname_ to _mat_
<2> Set the simple property _address.city_ to _emond's field_ without having to pass in the entire object. This does not work when a custom conversion is registered.
<3> Remove the property _age_.
<4> Set complex property _address_.
<5> Set a map/collection of values removes the previously existing map/collection and replaces the values with the given ones.
<6> Automatically update the server expiration time when altering time to live.
====

NOTE: Updating complex objects as well as map/collection structures requires further interaction with Redis to determine existing values which means that it might turn out that rewriting the entire entity might be faster.

Original pull request: #191.
---
 .../reference/redis-repositories.adoc         |  38 +
 .../data/redis/core/IndexWriter.java          |  17 +-
 .../data/redis/core/PartialUpdate.java        | 236 ++++++
 .../data/redis/core/RedisKeyValueAdapter.java | 149 ++++
 .../redis/core/RedisKeyValueTemplate.java     |  76 +-
 .../core/convert/CompositeIndexResolver.java  |  11 +
 .../redis/core/convert/IndexResolver.java     |   2 +
 .../core/convert/MappingRedisConverter.java   | 177 ++++-
 .../redis/core/convert/PathIndexResolver.java |  30 +-
 .../redis/core/convert/RedisConverter.java    |  11 +-
 .../data/redis/core/convert/RedisData.java    |  12 +-
 .../redis/core/convert/RemoveIndexedData.java |  53 ++
 .../redis/core/convert/SpelIndexResolver.java |   6 +
 .../core/mapping/RedisMappingContext.java     |  44 +-
 .../data/redis/hash/ObjectHashMapper.java     |  18 +
 .../redis/core/RedisKeyValueAdapterTests.java | 227 +++++-
 .../core/RedisKeyValueTemplateTests.java      | 676 +++++++++++++++++-
 .../MappingRedisConverterUnitTests.java       | 407 ++++++++++-
 .../convert/PathIndexResolverUnitTests.java   |   5 +-
 ...onfigAwareTimeToLiveAccessorUnitTests.java |  50 ++
 20 files changed, 2185 insertions(+), 60 deletions(-)
 create mode 100644 src/main/java/org/springframework/data/redis/core/PartialUpdate.java
 create mode 100644 src/main/java/org/springframework/data/redis/core/convert/RemoveIndexedData.java

diff --git a/src/main/asciidoc/reference/redis-repositories.adoc b/src/main/asciidoc/reference/redis-repositories.adoc
index f190b11b2..3e15ae3a1 100644
--- a/src/main/asciidoc/reference/redis-repositories.adoc
+++ b/src/main/asciidoc/reference/redis-repositories.adoc
@@ -493,6 +493,44 @@ mother = persons:a9d4b3a0-50d3-4538-a2fc-f7fc2581ee56      <1>
 WARNING: Referenced Objects are not subject of persisting changes when saving the referencing object. Please make sure to persist changes on referenced objects separately, since only the reference will be stored.
 Indexes set on properties of referenced types will not be resolved.
 
+[[redis.repositories.partial-updates]]
+== Persisting Partial Updates
+In some cases it is not necessary to load and rewrite the entire entity just to set a new value within it. A session timestamp for last active time might be such a scenario where you just want to alter one property.
+`PartialUpdate` allows to define `set`, `delete` actions on existing objects while taking care of updating potential expiration times of the entity itself as well as index structures.
+
+.Sample Partial Update
+====
+[source,java]
+----
+PartialUpdate<Person> update = new PartialUpdate<Person>("e2c7dcee", Person.class)
+  .set("firstname", "mat")                                                           <1>
+  .set("address.city", "emond's field")                                              <2>
+  .del("age");                                                                       <3>
+
+template.update(update);
+
+update = new PartialUpdate<Person>("e2c7dcee", Person.class)
+  .set("address", new Address("caemlyn", "andor"))                                   <4>
+  .set("attributes", singletonMap("eye-color", "grey"));                             <5>
+
+template.update(update);
+
+update = new PartialUpdate<Person>("e2c7dcee", Person.class)
+  .refreshTtl(true);                                                                 <6>
+  .set("expiration", 1000);
+
+template.update(update);
+----
+<1> Set the simple property _firstname_ to _mat_
+<2> Set the simple property _address.city_ to _emond's field_ without having to pass in the entire object. This does not work when a custom conversion is registered.
+<3> Remove the property _age_.
+<4> Set complex property _address_.
+<5> Set a map/collection of values removes the previously existing map/collection and replaces the values with the given ones.
+<6> Automatically update the server expiration time when altering <<redis.repositories.expirations>>.
+====
+
+NOTE: Updating complex objects as well as map/collection structures requires further interaction with Redis to determine existing values which means that it might turn out that rewriting the entire entity might be faster.
+
 [[redis.repositories.queries]]
 == Queries and Query Methods
 Query methods allow automatic derivation of simple finder queries from the method name.
diff --git a/src/main/java/org/springframework/data/redis/core/IndexWriter.java b/src/main/java/org/springframework/data/redis/core/IndexWriter.java
index e25a7a52e..952ca4b2b 100644
--- a/src/main/java/org/springframework/data/redis/core/IndexWriter.java
+++ b/src/main/java/org/springframework/data/redis/core/IndexWriter.java
@@ -21,6 +21,7 @@ import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.core.convert.IndexedData;
 import org.springframework.data.redis.core.convert.RedisConverter;
+import org.springframework.data.redis.core.convert.RemoveIndexedData;
 import org.springframework.data.redis.core.convert.SimpleIndexedPropertyValue;
 import org.springframework.data.redis.util.ByteUtils;
 import org.springframework.util.Assert;
@@ -32,7 +33,7 @@ import org.springframework.util.ObjectUtils;
  * Redis. Depending on the type of {@link IndexedData} it uses eg. Sets with specific names to add actually referenced
  * keys to. While doing so {@link IndexWriter} also keeps track of all indexes associated with the root types key, which
  * allows to remove the root key from all indexes in case of deletion.
- * 
+ *
  * @author Christoph Strobl
  * @author Rob Winch
  * @since 1.7
@@ -44,7 +45,7 @@ class IndexWriter {
 
 	/**
 	 * Creates new {@link IndexWriter}.
-	 * 
+	 *
 	 * @param keyspace The key space to write index values to. Must not be {@literal null}.
 	 * @param connection must not be {@literal null}.
 	 * @param converter must not be {@literal null}.
@@ -104,7 +105,7 @@ class IndexWriter {
 
 	/**
 	 * Removes a key from all available indexes.
-	 * 
+	 *
 	 * @param key must not be {@literal null}.
 	 */
 	public void removeKeyFromIndexes(String keyspace, Object key) {
@@ -142,7 +143,7 @@ class IndexWriter {
 
 	/**
 	 * Remove given key from all indexes matching {@link IndexedData#getIndexName()}:
-	 * 
+	 *
 	 * @param key
 	 * @param indexedData
 	 */
@@ -168,7 +169,7 @@ class IndexWriter {
 
 	/**
 	 * Adds a given key to the index for {@link IndexedData#getIndexName()}.
-	 * 
+	 *
 	 * @param key must not be {@literal null}.
 	 * @param indexedData must not be {@literal null}.
 	 */
@@ -177,7 +178,11 @@ class IndexWriter {
 		Assert.notNull(key, "Key must not be null!");
 		Assert.notNull(indexedData, "IndexedData must not be null!");
 
-		if (indexedData instanceof SimpleIndexedPropertyValue) {
+		if (indexedData instanceof RemoveIndexedData) {
+			return;
+		}
+
+		else if (indexedData instanceof SimpleIndexedPropertyValue) {
 
 			Object value = ((SimpleIndexedPropertyValue) indexedData).getValue();
 
diff --git a/src/main/java/org/springframework/data/redis/core/PartialUpdate.java b/src/main/java/org/springframework/data/redis/core/PartialUpdate.java
new file mode 100644
index 000000000..03aef53bc
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/core/PartialUpdate.java
@@ -0,0 +1,236 @@
+/*
+ * 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;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
+/**
+ * {@link PartialUpdate} allows to issue individual property updates without the need of rewriting the whole entity. It
+ * allows to define {@literal set}, {@literal delete} actions on existing objects while taking care of updating
+ * potential expiration times of the entity itself as well as index structures.
+ *
+ * @author Christoph Strobl
+ * @param <T>
+ * @since 1.8
+ */
+public class PartialUpdate<T> {
+
+	private final Object id;
+	private final Class<T> target;
+	private final T value;
+	private boolean refreshTtl = false;
+
+	private final List<PropertyUpdate> propertyUpdates = new ArrayList<PropertyUpdate>();
+
+	private PartialUpdate(Object id, Class<T> target, T value, boolean refreshTtl, List<PropertyUpdate> propertyUpdates) {
+
+		this.id = id;
+		this.target = target;
+		this.value = value;
+		this.refreshTtl = refreshTtl;
+		this.propertyUpdates.addAll(propertyUpdates);
+	}
+
+	/**
+	 * Create new {@link PartialUpdate} for given id and type.
+	 *
+	 * @param id must not be {@literal null}.
+	 * @param targetType must not be {@literal null}.
+	 */
+	@SuppressWarnings("unchecked")
+	public PartialUpdate(Object id, Class<T> targetType) {
+
+		Assert.notNull(id, "Id must not be null!");
+		Assert.notNull(targetType, "TargetType must not be null!");
+
+		this.id = id;
+		this.target = (Class<T>) ClassUtils.getUserClass(targetType);
+		this.value = null;
+	}
+
+	/**
+	 * Create new {@link PartialUpdate} for given id and object.
+	 *
+	 * @param id must not be {@literal null}.
+	 * @param value must not be {@literal null}.
+	 */
+	@SuppressWarnings("unchecked")
+	public PartialUpdate(Object id, T value) {
+
+		Assert.notNull(id, "Id must not be null!");
+		Assert.notNull(value, "Value must not be null!");
+
+		this.id = id;
+		this.target = (Class<T>) ClassUtils.getUserClass(value.getClass());
+		this.value = value;
+	}
+
+	/**
+	 * Create new {@link PartialUpdate} for given id and type.
+	 *
+	 * @param id must not be {@literal null}.
+	 * @param targetType must not be {@literal null}.
+	 */
+	public static <S> PartialUpdate<S> newPartialUpdate(Object id, Class<S> targetType) {
+		return new PartialUpdate<S>(id, targetType);
+	}
+
+	/**
+	 * @return can be {@literal null}.
+	 */
+	public T getValue() {
+		return value;
+	}
+
+	/**
+	 * Set the value of a simple or complex {@literal value} reachable via given {@literal path}.
+	 *
+	 * @param path must not be {@literal null}.
+	 * @param value must not be {@literal null}. If you want to remove a value use {@link #del(String)}.
+	 * @return a new {@link PartialUpdate}.
+	 */
+	public PartialUpdate<T> set(String path, Object value) {
+
+		Assert.hasText(path, "Path to set must not be null or empty!");
+
+		PartialUpdate<T> update = new PartialUpdate<T>(this.id, this.target, this.value, this.refreshTtl,
+				this.propertyUpdates);
+		update.propertyUpdates.add(new PropertyUpdate(UpdateCommand.SET, path, value));
+		return update;
+	}
+
+	/**
+	 * Remove the value reachable via given {@literal path}.
+	 *
+	 * @param path path must not be {@literal null}.
+	 * @return a new {@link PartialUpdate}.
+	 */
+	public PartialUpdate<T> del(String path) {
+
+		Assert.hasText(path, "Path to remove must not be null or empty!");
+
+		PartialUpdate<T> update = new PartialUpdate<T>(this.id, this.target, this.value, this.refreshTtl,
+				this.propertyUpdates);
+		update.propertyUpdates.add(new PropertyUpdate(UpdateCommand.DEL, path));
+		return update;
+	}
+
+	/**
+	 * Get the target type.
+	 *
+	 * @return never {@literal null}.
+	 */
+	public Class<T> getTarget() {
+		return target;
+	}
+
+	/**
+	 * Get the id of the element to update.
+	 *
+	 * @return never {@literal null}.
+	 */
+	public Object getId() {
+		return id;
+	}
+
+	/**
+	 * Get the list of individual property updates.
+	 *
+	 * @return never {@literal null}.
+	 */
+	public List<PropertyUpdate> getPropertyUpdates() {
+		return Collections.unmodifiableList(propertyUpdates);
+	}
+
+	/**
+	 * @return true if expiration time of target should be updated.
+	 */
+	public boolean isRefreshTtl() {
+		return refreshTtl;
+	}
+
+	/**
+	 * Set indicator for updating expiration time of target.
+	 *
+	 * @param refreshTtl
+	 * @return a new {@link PartialUpdate}.
+	 */
+	public PartialUpdate<T> refreshTtl(boolean refreshTtl) {
+		return new PartialUpdate<T>(this.id, this.target, this.value, refreshTtl, this.propertyUpdates);
+	}
+
+	/**
+	 * @author Christoph Strobl
+	 * @since 1.8
+	 */
+	public static class PropertyUpdate {
+
+		private final UpdateCommand cmd;
+		private final String propertyPath;
+		private final Object value;
+
+		private PropertyUpdate(UpdateCommand cmd, String propertyPath) {
+			this(cmd, propertyPath, null);
+		}
+
+		private PropertyUpdate(UpdateCommand cmd, String propertyPath, Object value) {
+
+			this.cmd = cmd;
+			this.propertyPath = propertyPath;
+			this.value = value;
+		}
+
+		/**
+		 * Get the associated {@link UpdateCommand}.
+		 *
+		 * @return never {@literal null}.
+		 */
+		public UpdateCommand getCmd() {
+			return cmd;
+		}
+
+		/**
+		 * Get the target path.
+		 *
+		 * @return never {@literal null}.
+		 */
+		public String getPropertyPath() {
+			return propertyPath;
+		}
+
+		/**
+		 * Get the value to set.
+		 *
+		 * @return can be {@literal null}.
+		 */
+		public Object getValue() {
+			return value;
+		}
+	}
+
+	/**
+	 * @author Christoph Strobl
+	 * @since 1.8
+	 */
+	public static enum UpdateCommand {
+		SET, DEL
+	}
+}
diff --git a/src/main/java/org/springframework/data/redis/core/RedisKeyValueAdapter.java b/src/main/java/org/springframework/data/redis/core/RedisKeyValueAdapter.java
index 13ddf4896..b50b8250e 100644
--- a/src/main/java/org/springframework/data/redis/core/RedisKeyValueAdapter.java
+++ b/src/main/java/org/springframework/data/redis/core/RedisKeyValueAdapter.java
@@ -17,6 +17,8 @@ package org.springframework.data.redis.core;
 
 import java.io.Serializable;
 import java.util.ArrayList;
+import java.util.Collection;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -41,6 +43,8 @@ import org.springframework.data.keyvalue.core.mapping.KeyValuePersistentProperty
 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.KeyspaceConfiguration;
 import org.springframework.data.redis.core.convert.MappingRedisConverter;
@@ -49,6 +53,7 @@ import org.springframework.data.redis.core.convert.RedisConverter;
 import org.springframework.data.redis.core.convert.RedisData;
 import org.springframework.data.redis.core.convert.ReferenceResolverImpl;
 import org.springframework.data.redis.core.mapping.RedisMappingContext;
+import org.springframework.data.redis.core.mapping.RedisPersistentEntity;
 import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
 import org.springframework.data.redis.listener.RedisMessageListenerContainer;
 import org.springframework.data.redis.util.ByteUtils;
@@ -410,6 +415,121 @@ public class RedisKeyValueAdapter extends AbstractKeyValueAdapter
 		return count != null ? count.longValue() : 0;
 	}
 
+	public void update(final PartialUpdate<?> update) {
+
+		final RedisPersistentEntity<?> entity = this.converter.getMappingContext().getPersistentEntity(update.getTarget());
+
+		final String keyspace = entity.getKeySpace();
+		final Object id = update.getId();
+
+		final byte[] redisKey = createKey(keyspace, converter.getConversionService().convert(id, String.class));
+
+		final RedisData rdo = new RedisData();
+		this.converter.write(update, rdo);
+
+		redisOps.execute(new RedisCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisConnection connection) throws DataAccessException {
+
+				RedisUpdateObject redisUpdateObject = new RedisUpdateObject(redisKey, keyspace, id);
+
+				for (PropertyUpdate pUpdate : update.getPropertyUpdates()) {
+
+					String propertyPath = pUpdate.getPropertyPath();
+
+					if (UpdateCommand.DEL.equals(pUpdate.getCmd()) || pUpdate.getValue() instanceof Collection
+							|| pUpdate.getValue() instanceof Map
+							|| (pUpdate.getValue() != null && pUpdate.getValue().getClass().isArray()) || (pUpdate.getValue() != null
+									&& !converter.getConversionService().canConvert(pUpdate.getValue().getClass(), byte[].class))) {
+
+						redisUpdateObject = fetchDeletePathsFromHashAndUpdateIndex(redisUpdateObject, propertyPath, connection);
+					}
+				}
+
+				if (!redisUpdateObject.fieldsToRemove.isEmpty()) {
+					connection.hDel(redisKey,
+							redisUpdateObject.fieldsToRemove.toArray(new byte[redisUpdateObject.fieldsToRemove.size()][]));
+				}
+
+				for (byte[] index : redisUpdateObject.indexesToUpdate) {
+					connection.sRem(index, toBytes(redisUpdateObject.targetId));
+				}
+
+				if (!rdo.getBucket().isEmpty()) {
+					if (rdo.getBucket().size() > 1
+							|| (rdo.getBucket().size() == 1 && !rdo.getBucket().asMap().containsKey("_class"))) {
+						connection.hMSet(redisKey, rdo.getBucket().rawMap());
+					}
+				}
+
+				if (update.isRefreshTtl()) {
+
+					if (rdo.getTimeToLive() != null && rdo.getTimeToLive().longValue() > 0) {
+
+						connection.expire(redisKey, rdo.getTimeToLive().longValue());
+
+						// add phantom key so values can be restored
+						byte[] phantomKey = ByteUtils.concat(redisKey, toBytes(":phantom"));
+						connection.hMSet(phantomKey, rdo.getBucket().rawMap());
+						connection.expire(phantomKey, rdo.getTimeToLive().longValue() + 300);
+
+					} else {
+
+						connection.persist(redisKey);
+						connection.persist(ByteUtils.concat(redisKey, toBytes(":phantom")));
+					}
+				}
+
+				new IndexWriter(connection, converter).updateIndexes(toBytes(id), rdo.getIndexedData());
+				return null;
+			}
+
+		});
+	}
+
+	private RedisUpdateObject fetchDeletePathsFromHashAndUpdateIndex(RedisUpdateObject redisUpdateObject, String path,
+			RedisConnection connection) {
+
+		redisUpdateObject.addFieldToRemove(toBytes(path));
+		byte[] value = connection.hGet(redisUpdateObject.targetKey, toBytes(path));
+
+		if (value != null && value.length > 0) {
+
+			byte[] existingValueIndexKey = value != null
+					? ByteUtils.concatAll(toBytes(redisUpdateObject.keyspace), toBytes((":" + path)), toBytes(":"), value) : null;
+
+			if (connection.exists(existingValueIndexKey)) {
+				redisUpdateObject.addIndexToUpdate(existingValueIndexKey);
+			}
+			return redisUpdateObject;
+		}
+
+		Set<byte[]> existingFields = connection.hKeys(redisUpdateObject.targetKey);
+
+		for (byte[] field : existingFields) {
+
+			if (asString(field).startsWith(path + ".")) {
+
+				redisUpdateObject.addFieldToRemove(field);
+				value = connection.hGet(redisUpdateObject.targetKey, toBytes(field));
+
+				if (value != null) {
+
+					byte[] existingValueIndexKey = value != null
+							? ByteUtils.concatAll(toBytes(redisUpdateObject.keyspace), toBytes(":"), field, toBytes(":"), value)
+							: null;
+
+					if (connection.exists(existingValueIndexKey)) {
+						redisUpdateObject.addIndexToUpdate(existingValueIndexKey);
+					}
+				}
+			}
+		}
+
+		return redisUpdateObject;
+	}
+
 	/**
 	 * Execute {@link RedisCallback} via underlying {@link RedisOperations}.
 	 *
@@ -659,4 +779,33 @@ public class RedisKeyValueAdapter extends AbstractKeyValueAdapter
 		OFF
 	}
 
+	/**
+	 * Container holding update information like fields to remove from the Redis Hash.
+	 *
+	 * @author Christoph Strobl
+	 */
+	private static class RedisUpdateObject {
+
+		private final String keyspace;
+		private final Object targetId;
+		private final byte[] targetKey;
+
+		private Set<byte[]> fieldsToRemove = new LinkedHashSet<byte[]>();
+		private Set<byte[]> indexesToUpdate = new LinkedHashSet<byte[]>();
+
+		RedisUpdateObject(byte[] targetKey, String keyspace, Object targetId) {
+
+			this.targetKey = targetKey;
+			this.keyspace = keyspace;
+			this.targetId = targetId;
+		}
+
+		void addFieldToRemove(byte[] field) {
+			fieldsToRemove.add(field);
+		}
+
+		void addIndexToUpdate(byte[] indexName) {
+			indexesToUpdate.add(indexName);
+		}
+	}
 }
diff --git a/src/main/java/org/springframework/data/redis/core/RedisKeyValueTemplate.java b/src/main/java/org/springframework/data/redis/core/RedisKeyValueTemplate.java
index 2f1d1f057..d75f70017 100644
--- a/src/main/java/org/springframework/data/redis/core/RedisKeyValueTemplate.java
+++ b/src/main/java/org/springframework/data/redis/core/RedisKeyValueTemplate.java
@@ -15,6 +15,7 @@
  */
 package org.springframework.data.redis.core;
 
+import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -22,6 +23,7 @@ import java.util.List;
 import org.springframework.data.keyvalue.core.KeyValueAdapter;
 import org.springframework.data.keyvalue.core.KeyValueCallback;
 import org.springframework.data.keyvalue.core.KeyValueTemplate;
+import org.springframework.data.redis.RedisSystemException;
 import org.springframework.data.redis.core.mapping.RedisMappingContext;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
@@ -70,6 +72,7 @@ public class RedisKeyValueTemplate extends KeyValueTemplate {
 	 * </code>
 	 * 
 	 * <pre>
+	 *
 	 * @param callback provides the to retrieve entity ids. Must not be {@literal null}.
 	 * @param type must not be {@literal null}.
 	 * @return empty list if not elements found.
@@ -89,15 +92,14 @@ public class RedisKeyValueTemplate extends KeyValueTemplate {
 					return Collections.emptyList();
 				}
 
-				Iterable<?> ids = ClassUtils.isAssignable(Iterable.class, callbackResult.getClass()) ? (Iterable<?>) callbackResult
-						: Collections.singleton(callbackResult);
+				Iterable<?> ids = ClassUtils.isAssignable(Iterable.class, callbackResult.getClass())
+						? (Iterable<?>) callbackResult : Collections.singleton(callbackResult);
 
 				List<T> result = new ArrayList<T>();
 				for (Object id : ids) {
 
-					String idToUse = adapter.getConverter().getConversionService().canConvert(id.getClass(), String.class) ? adapter
-							.getConverter().getConversionService().convert(id, String.class)
-							: id.toString();
+					String idToUse = adapter.getConverter().getConversionService().canConvert(id.getClass(), String.class)
+							? adapter.getConverter().getConversionService().convert(id, String.class) : id.toString();
 
 					T candidate = findById(idToUse, type);
 					if (candidate != null) {
@@ -110,6 +112,70 @@ public class RedisKeyValueTemplate extends KeyValueTemplate {
 		});
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.keyvalue.core.KeyValueTemplate#insert(java.io.Serializable, java.lang.Object)
+	 */
+	@Override
+	public void insert(final Serializable id, final Object objectToInsert) {
+
+		if (objectToInsert instanceof PartialUpdate) {
+			doPartialUpdate((PartialUpdate<?>) objectToInsert);
+		}
+
+		super.insert(id, objectToInsert);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.keyvalue.core.KeyValueTemplate#update(java.lang.Object)
+	 */
+	@Override
+	public void update(Object objectToUpdate) {
+
+		if (objectToUpdate instanceof PartialUpdate) {
+			doPartialUpdate((PartialUpdate<?>) objectToUpdate);
+		}
+
+		super.update(objectToUpdate);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.keyvalue.core.KeyValueTemplate#destroy()
+	 */
+	@Override
+	public void destroy() throws Exception {
+
+		execute(new RedisKeyValueCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisKeyValueAdapter adapter) {
+
+				try {
+					adapter.destroy();
+				} catch (Exception e) {
+					throw new RedisSystemException(e.getMessage(), e);
+				}
+				return null;
+			}
+		});
+
+	}
+
+	protected void doPartialUpdate(final PartialUpdate<?> update) {
+
+		execute(new RedisKeyValueCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisKeyValueAdapter adapter) {
+
+				adapter.update(update);
+				return null;
+			}
+		});
+	}
+
 	/**
 	 * Redis specific {@link KeyValueCallback}.
 	 * 
diff --git a/src/main/java/org/springframework/data/redis/core/convert/CompositeIndexResolver.java b/src/main/java/org/springframework/data/redis/core/convert/CompositeIndexResolver.java
index 2ebe21b3f..15b0e2de6 100644
--- a/src/main/java/org/springframework/data/redis/core/convert/CompositeIndexResolver.java
+++ b/src/main/java/org/springframework/data/redis/core/convert/CompositeIndexResolver.java
@@ -72,4 +72,15 @@ public class CompositeIndexResolver implements IndexResolver {
 		return data;
 	}
 
+	@Override
+	public Set<IndexedData> resolveIndexesFor(String keyspace, String path, TypeInformation<?> typeInformation,
+			Object value) {
+
+		Set<IndexedData> data = new LinkedHashSet<IndexedData>();
+		for (IndexResolver resolver : resolvers) {
+			data.addAll(resolver.resolveIndexesFor(keyspace, path, typeInformation, value));
+		}
+		return data;
+	}
+
 }
diff --git a/src/main/java/org/springframework/data/redis/core/convert/IndexResolver.java b/src/main/java/org/springframework/data/redis/core/convert/IndexResolver.java
index a90e376f7..08e39f6e9 100644
--- a/src/main/java/org/springframework/data/redis/core/convert/IndexResolver.java
+++ b/src/main/java/org/springframework/data/redis/core/convert/IndexResolver.java
@@ -38,4 +38,6 @@ public interface IndexResolver {
 	 */
 	Set<IndexedData> resolveIndexesFor(TypeInformation<?> typeInformation, Object value);
 
+	Set<IndexedData> resolveIndexesFor(String keyspace, String path, TypeInformation<?> typeInformation, Object value);
+
 }
diff --git a/src/main/java/org/springframework/data/redis/core/convert/MappingRedisConverter.java b/src/main/java/org/springframework/data/redis/core/convert/MappingRedisConverter.java
index c503e71fe..94ab1e0d1 100644
--- a/src/main/java/org/springframework/data/redis/core/convert/MappingRedisConverter.java
+++ b/src/main/java/org/springframework/data/redis/core/convert/MappingRedisConverter.java
@@ -48,9 +48,13 @@ import org.springframework.data.mapping.AssociationHandler;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
 import org.springframework.data.mapping.PreferredConstructor;
 import org.springframework.data.mapping.PropertyHandler;
+import org.springframework.data.mapping.context.PersistentPropertyPath;
 import org.springframework.data.mapping.model.MappingException;
 import org.springframework.data.mapping.model.PersistentEntityParameterValueProvider;
 import org.springframework.data.mapping.model.PropertyValueProvider;
+import org.springframework.data.redis.core.PartialUpdate;
+import org.springframework.data.redis.core.PartialUpdate.PropertyUpdate;
+import org.springframework.data.redis.core.PartialUpdate.UpdateCommand;
 import org.springframework.data.redis.core.index.Indexed;
 import org.springframework.data.redis.core.mapping.RedisMappingContext;
 import org.springframework.data.redis.core.mapping.RedisPersistentEntity;
@@ -109,6 +113,7 @@ import org.springframework.util.comparator.NullSafeComparator;
 public class MappingRedisConverter implements RedisConverter, InitializingBean {
 
 	private static final String TYPE_HINT_ALIAS = "_class";
+	private static final String INVALID_TYPE_ASSIGNMENT = "Value of type %s cannot be assigned to property %s of type %s.";
 
 	private final RedisMappingContext mappingContext;
 	private final GenericConversionService conversionService;
@@ -223,12 +228,18 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 
 				if (persistentProperty.isMap()) {
 
+					Map<?, ?> targetValue = null;
+
 					if (conversionService.canConvert(byte[].class, persistentProperty.getMapValueType())) {
-						accessor.setProperty(persistentProperty, readMapOfSimpleTypes(currentPath, persistentProperty.getType(),
-								persistentProperty.getComponentType(), persistentProperty.getMapValueType(), source));
+						targetValue = readMapOfSimpleTypes(currentPath, persistentProperty.getType(),
+								persistentProperty.getComponentType(), persistentProperty.getMapValueType(), source);
 					} else {
-						accessor.setProperty(persistentProperty, readMapOfComplexTypes(currentPath, persistentProperty.getType(),
-								persistentProperty.getComponentType(), persistentProperty.getMapValueType(), source));
+						targetValue = readMapOfComplexTypes(currentPath, persistentProperty.getType(),
+								persistentProperty.getComponentType(), persistentProperty.getMapValueType(), source);
+					}
+
+					if (targetValue != null) {
+						accessor.setProperty(persistentProperty, targetValue);
 					}
 				}
 
@@ -236,7 +247,9 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 
 					Object targetValue = readCollectionOrArray(currentPath, persistentProperty.getType(),
 							persistentProperty.getTypeInformation().getComponentType().getActualType().getType(), source.getBucket());
-					accessor.setProperty(persistentProperty, targetValue);
+					if (targetValue != null) {
+						accessor.setProperty(persistentProperty, targetValue);
+					}
 
 				} else if (persistentProperty.isEntity() && !conversionService.canConvert(byte[].class,
 						persistentProperty.getTypeInformation().getActualType().getType())) {
@@ -339,6 +352,15 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 	@SuppressWarnings({ "rawtypes" })
 	public void write(Object source, final RedisData sink) {
 
+		if (source == null) {
+			return;
+		}
+
+		if (source instanceof PartialUpdate) {
+			writePartialUpdate((PartialUpdate) source, sink);
+			return;
+		}
+
 		final RedisPersistentEntity entity = mappingContext.getPersistentEntity(source.getClass());
 
 		if (!customConversions.hasCustomWriteTarget(source.getClass())) {
@@ -365,7 +387,125 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 		for (IndexedData indexedData : indexResolver.resolveIndexesFor(entity.getTypeInformation(), source)) {
 			sink.addIndexedData(indexedData);
 		}
+	}
 
+	protected void writePartialUpdate(PartialUpdate<?> update, RedisData sink) {
+
+		RedisPersistentEntity<?> entity = mappingContext.getPersistentEntity(update.getTarget());
+
+		write(update.getValue(), sink);
+		if (sink.getBucket().keySet().contains("_class")) {
+			sink.getBucket().put("_class", null); // overwrite stuff in here
+		}
+
+		if (update.isRefreshTtl() && !update.getPropertyUpdates().isEmpty()) {
+
+			Long ttl = entity.getTimeToLiveAccessor().getTimeToLive(update);
+			if (ttl != null && ttl > 0) {
+				sink.setTimeToLive(ttl);
+			}
+		}
+
+		for (PropertyUpdate pUpdate : update.getPropertyUpdates()) {
+
+			String path = pUpdate.getPropertyPath();
+
+			if (UpdateCommand.SET.equals(pUpdate.getCmd())) {
+
+				KeyValuePersistentProperty targetProperty = getTargetPropertyOrNullForPath(path, update.getTarget());
+
+				if (targetProperty == null) {
+
+					targetProperty = getTargetPropertyOrNullForPath(path.replaceAll("\\.\\[.*\\]", ""), update.getTarget());
+
+					TypeInformation<?> ti = targetProperty == null ? ClassTypeInformation.OBJECT
+							: (targetProperty.isMap()
+									? (targetProperty.getTypeInformation().getMapValueType() != null
+											? targetProperty.getTypeInformation().getMapValueType() : ClassTypeInformation.OBJECT)
+									: targetProperty.getTypeInformation().getActualType());
+
+					writeInternal(entity.getKeySpace(), pUpdate.getPropertyPath(), pUpdate.getValue(), ti, sink);
+					continue;
+				}
+
+				if (targetProperty.isAssociation()) {
+
+					if (targetProperty.isCollectionLike()) {
+
+						KeyValuePersistentEntity<?> ref = mappingContext.getPersistentEntity(
+								targetProperty.getAssociation().getInverse().getTypeInformation().getComponentType().getActualType());
+
+						int i = 0;
+						for (Object o : (Collection<?>) pUpdate.getValue()) {
+
+							Object refId = ref.getPropertyAccessor(o).getProperty(ref.getIdProperty());
+							sink.getBucket().put(pUpdate.getPropertyPath() + ".[" + i + "]",
+									toBytes(ref.getKeySpace() + ":" + refId));
+							i++;
+						}
+					} else {
+
+						KeyValuePersistentEntity<?> ref = mappingContext
+								.getPersistentEntity(targetProperty.getAssociation().getInverse().getTypeInformation());
+
+						Object refId = ref.getPropertyAccessor(pUpdate.getValue()).getProperty(ref.getIdProperty());
+						sink.getBucket().put(pUpdate.getPropertyPath(), toBytes(ref.getKeySpace() + ":" + refId));
+					}
+				}
+
+				else if (targetProperty.isCollectionLike()) {
+
+					Collection<?> collection = pUpdate.getValue() instanceof Collection ? (Collection<?>) pUpdate.getValue()
+							: Collections.<Object> singleton(pUpdate.getValue());
+					writeCollection(entity.getKeySpace(), pUpdate.getPropertyPath(), collection,
+							targetProperty.getTypeInformation().getActualType(), sink);
+				} else if (targetProperty.isMap()) {
+
+					Map<Object, Object> map = new HashMap<Object, Object>();
+
+					if (pUpdate.getValue() instanceof Map) {
+						map.putAll((Map<?, ?>) pUpdate.getValue());
+					} else if (pUpdate.getValue() instanceof Map.Entry) {
+						map.put(((Map.Entry<?, ?>) pUpdate.getValue()).getKey(), ((Map.Entry<?, ?>) pUpdate.getValue()).getValue());
+					} else {
+						throw new MappingException(
+								String.format("Cannot set update value for map property '%s' to '%s'. Please use a Map or Map.Entry.",
+										pUpdate.getPropertyPath(), pUpdate.getValue()));
+					}
+
+					writeMap(entity.getKeySpace(), pUpdate.getPropertyPath(), targetProperty.getMapValueType(), map, sink);
+				} else {
+
+					writeInternal(entity.getKeySpace(), pUpdate.getPropertyPath(), pUpdate.getValue(),
+							targetProperty.getTypeInformation(), sink);
+
+					Set<IndexedData> data = indexResolver.resolveIndexesFor(entity.getKeySpace(), pUpdate.getPropertyPath(),
+							targetProperty.getTypeInformation(), pUpdate.getValue());
+
+					if (data.isEmpty()) {
+
+						data = indexResolver.resolveIndexesFor(entity.getKeySpace(), pUpdate.getPropertyPath(),
+								targetProperty.getOwner().getTypeInformation(), pUpdate.getValue());
+
+					}
+					sink.addIndexedData(data);
+				}
+			}
+		}
+	}
+
+	KeyValuePersistentProperty getTargetPropertyOrNullForPath(String path, Class<?> type) {
+
+		try {
+
+			PersistentPropertyPath<KeyValuePersistentProperty> persistentPropertyPath = mappingContext
+					.getPersistentPropertyPath(path, type);
+			return persistentPropertyPath.getLeafProperty();
+		} catch (Exception e) {
+			// that's just fine
+		}
+
+		return null;
 	}
 
 	/**
@@ -387,6 +527,11 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 			if (!StringUtils.hasText(path) && customConversions.getCustomWriteTarget(value.getClass()).equals(byte[].class)) {
 				sink.getBucket().put(StringUtils.hasText(path) ? path : "_raw", conversionService.convert(value, byte[].class));
 			} else {
+
+				if (!ClassUtils.isAssignable(typeHint.getType(), value.getClass())) {
+					throw new MappingException(
+							String.format(INVALID_TYPE_ASSIGNMENT, value.getClass(), path, typeHint.getType()));
+				}
 				writeToBucket(path, value, sink, typeHint.getType());
 			}
 			return;
@@ -520,6 +665,11 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 
 			String currentPath = path + ".[" + i + "]";
 
+			if (!ClassUtils.isAssignable(typeHint.getType(), value.getClass())) {
+				throw new MappingException(
+						String.format(INVALID_TYPE_ASSIGNMENT, value.getClass(), currentPath, typeHint.getType()));
+			}
+
 			if (customConversions.hasCustomWriteTarget(value.getClass())) {
 				writeToBucket(currentPath, value, sink, typeHint.getType());
 			} else {
@@ -592,7 +742,7 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 			}
 		}
 
-		return isArray ? toArray(target, collectionType, valueType) : target;
+		return isArray ? toArray(target, collectionType, valueType) : (target.isEmpty() ? null : target);
 	}
 
 	/**
@@ -616,6 +766,11 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 
 			String currentPath = path + ".[" + entry.getKey() + "]";
 
+			if (!ClassUtils.isAssignable(mapValueType, entry.getValue().getClass())) {
+				throw new MappingException(
+						String.format(INVALID_TYPE_ASSIGNMENT, entry.getValue().getClass(), currentPath, mapValueType));
+			}
+
 			if (customConversions.hasCustomWriteTarget(entry.getValue().getClass())) {
 				writeToBucket(currentPath, entry.getValue(), sink, mapValueType);
 			} else {
@@ -659,7 +814,7 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 			target.put(key, fromBytes(entry.getValue(), typeToUse));
 		}
 
-		return target;
+		return target.isEmpty() ? null : target;
 	}
 
 	/**
@@ -700,7 +855,7 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 			target.put(mapKey, o);
 		}
 
-		return target;
+		return target.isEmpty() ? null : target;
 	}
 
 	private Class<?> getTypeHint(String path, Bucket bucket, Class<?> fallback) {
@@ -759,6 +914,10 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 	 */
 	private Object toArray(Collection<Object> source, Class<?> arrayType, Class<?> valueType) {
 
+		if (source.isEmpty()) {
+			return null;
+		}
+
 		if (!ClassUtils.isPrimitiveArray(arrayType)) {
 			return source.toArray((Object[]) Array.newInstance(valueType, source.size()));
 		}
@@ -770,7 +929,7 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 			Array.set(targetArray, i, conversionService.convert(iterator.next(), valueType));
 			i++;
 		}
-		return targetArray;
+		return i > 0 ? targetArray : null;
 	}
 
 	/**
diff --git a/src/main/java/org/springframework/data/redis/core/convert/PathIndexResolver.java b/src/main/java/org/springframework/data/redis/core/convert/PathIndexResolver.java
index eeaf6a931..034f155ab 100644
--- a/src/main/java/org/springframework/data/redis/core/convert/PathIndexResolver.java
+++ b/src/main/java/org/springframework/data/redis/core/convert/PathIndexResolver.java
@@ -15,7 +15,6 @@
  */
 package org.springframework.data.redis.core.convert;
 
-import java.util.Collections;
 import java.util.LinkedHashSet;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -37,6 +36,7 @@ import org.springframework.data.redis.core.mapping.RedisPersistentEntity;
 import org.springframework.data.util.ClassTypeInformation;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
 import org.springframework.util.CollectionUtils;
 
 /**
@@ -80,6 +80,12 @@ public class PathIndexResolver implements IndexResolver {
 				null, value);
 	}
 
+	@Override
+	public Set<IndexedData> resolveIndexesFor(String keyspace, String path, TypeInformation<?> typeInformation,
+			Object value) {
+		return doResolveIndexesFor(keyspace, path, typeInformation, null, value);
+	}
+
 	private Set<IndexedData> doResolveIndexesFor(final String keyspace, final String path,
 			TypeInformation<?> typeInformation, PersistentProperty<?> fallback, Object value) {
 
@@ -89,6 +95,13 @@ public class PathIndexResolver implements IndexResolver {
 			return resolveIndex(keyspace, path, fallback, value);
 		}
 
+		// this might happen on update where we address a property within an entity directly
+		if (!ClassUtils.isAssignable(entity.getType(), value.getClass())) {
+
+			String propertyName = path.lastIndexOf('.') > 0 ? path.substring(path.lastIndexOf('.') + 1, path.length()) : path;
+			return resolveIndex(keyspace, path, entity.getPersistentProperty(propertyName), value);
+		}
+
 		final PersistentPropertyAccessor accessor = entity.getPropertyAccessor(value);
 		final Set<IndexedData> indexes = new LinkedHashSet<IndexedData>();
 
@@ -173,10 +186,6 @@ public class PathIndexResolver implements IndexResolver {
 	protected Set<IndexedData> resolveIndex(String keyspace, String propertyPath, PersistentProperty<?> property,
 			Object value) {
 
-		if (value == null) {
-			return Collections.emptySet();
-		}
-
 		String path = normalizeIndexPath(propertyPath, property);
 
 		Set<IndexedData> data = new LinkedHashSet<IndexedData>();
@@ -192,8 +201,14 @@ public class PathIndexResolver implements IndexResolver {
 					continue;
 				}
 
-				data.add(new SimpleIndexedPropertyValue(keyspace, indexDefinition.getIndexName(),
-						indexDefinition.valueTransformer().convert(value)));
+				Object transformedValue = indexDefinition.valueTransformer().convert(value);
+
+				IndexedData indexedData = new SimpleIndexedPropertyValue(keyspace, indexDefinition.getIndexName(),
+						transformedValue);
+				if (transformedValue == null) {
+					indexedData = new RemoveIndexedData(indexedData);
+				}
+				data.add(indexedData);
 			}
 		}
 
@@ -204,6 +219,7 @@ public class PathIndexResolver implements IndexResolver {
 
 			data.add(new SimpleIndexedPropertyValue(keyspace, path, indexDefinition.valueTransformer().convert(value)));
 		}
+
 		return data;
 	}
 
diff --git a/src/main/java/org/springframework/data/redis/core/convert/RedisConverter.java b/src/main/java/org/springframework/data/redis/core/convert/RedisConverter.java
index 5754da8a8..d8c94eec4 100644
--- a/src/main/java/org/springframework/data/redis/core/convert/RedisConverter.java
+++ b/src/main/java/org/springframework/data/redis/core/convert/RedisConverter.java
@@ -18,6 +18,7 @@ package org.springframework.data.redis.core.convert;
 import org.springframework.data.convert.EntityConverter;
 import org.springframework.data.keyvalue.core.mapping.KeyValuePersistentEntity;
 import org.springframework.data.keyvalue.core.mapping.KeyValuePersistentProperty;
+import org.springframework.data.redis.core.mapping.RedisMappingContext;
 
 /**
  * Redis specific {@link EntityConverter}.
@@ -25,7 +26,13 @@ import org.springframework.data.keyvalue.core.mapping.KeyValuePersistentProperty
  * @author Christoph Strobl
  * @since 1.7
  */
-public interface RedisConverter extends
-		EntityConverter<KeyValuePersistentEntity<?>, KeyValuePersistentProperty, Object, RedisData> {
+public interface RedisConverter
+		extends EntityConverter<KeyValuePersistentEntity<?>, KeyValuePersistentProperty, Object, RedisData> {
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.convert.EntityConverter#getMappingContext()
+	 */
+	@Override
+	RedisMappingContext getMappingContext();
 }
diff --git a/src/main/java/org/springframework/data/redis/core/convert/RedisData.java b/src/main/java/org/springframework/data/redis/core/convert/RedisData.java
index 0bfd78b6c..03e8434d5 100644
--- a/src/main/java/org/springframework/data/redis/core/convert/RedisData.java
+++ b/src/main/java/org/springframework/data/redis/core/convert/RedisData.java
@@ -15,6 +15,7 @@
  */
 package org.springframework.data.redis.core.convert;
 
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Map;
@@ -94,7 +95,7 @@ public class RedisData {
 	}
 
 	/**
-	 * @param index
+	 * @param index must not be {@literal null}.
 	 */
 	public void addIndexedData(IndexedData index) {
 
@@ -102,6 +103,15 @@ public class RedisData {
 		this.indexedData.add(index);
 	}
 
+	/**
+	 * @param indexes must not be {@literal null}.
+	 */
+	public void addIndexedData(Collection<IndexedData> indexes) {
+
+		Assert.notNull(indexes, "IndexedData to add must not be null!");
+		this.indexedData.addAll(indexes);
+	}
+
 	/**
 	 * @return never {@literal null}.
 	 */
diff --git a/src/main/java/org/springframework/data/redis/core/convert/RemoveIndexedData.java b/src/main/java/org/springframework/data/redis/core/convert/RemoveIndexedData.java
new file mode 100644
index 000000000..4030b8e6f
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/core/convert/RemoveIndexedData.java
@@ -0,0 +1,53 @@
+/*
+ * 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;
+
+/**
+ * @author Christoph Strobl
+ */
+public class RemoveIndexedData implements IndexedData {
+
+	private final IndexedData delegate;
+
+	RemoveIndexedData(IndexedData delegate) {
+		super();
+		this.delegate = delegate;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.convert.IndexedData#getIndexName()
+	 */
+	@Override
+	public String getIndexName() {
+		return delegate.getIndexName();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.convert.IndexedData#getKeyspace()
+	 */
+	@Override
+	public String getKeyspace() {
+		return delegate.getKeyspace();
+	}
+
+	@Override
+	public String toString() {
+		return "RemoveIndexedData [indexName=" + getIndexName() + ", keyspace()=" + getKeyspace() + "]";
+	}
+
+}
diff --git a/src/main/java/org/springframework/data/redis/core/convert/SpelIndexResolver.java b/src/main/java/org/springframework/data/redis/core/convert/SpelIndexResolver.java
index b4203f243..60f24ebf7 100644
--- a/src/main/java/org/springframework/data/redis/core/convert/SpelIndexResolver.java
+++ b/src/main/java/org/springframework/data/redis/core/convert/SpelIndexResolver.java
@@ -119,6 +119,12 @@ public class SpelIndexResolver implements IndexResolver {
 		return indexes;
 	}
 
+	@Override
+	public Set<IndexedData> resolveIndexesFor(String keyspace, String path, TypeInformation<?> typeInformation,
+			Object value) {
+		return Collections.emptySet();
+	}
+
 	private Expression getAndCacheIfAbsent(SpelIndexDefinition indexDefinition) {
 
 		if (expressionCache.containsKey(indexDefinition)) {
diff --git a/src/main/java/org/springframework/data/redis/core/mapping/RedisMappingContext.java b/src/main/java/org/springframework/data/redis/core/mapping/RedisMappingContext.java
index 3017bda5f..f4c333fd2 100644
--- a/src/main/java/org/springframework/data/redis/core/mapping/RedisMappingContext.java
+++ b/src/main/java/org/springframework/data/redis/core/mapping/RedisMappingContext.java
@@ -32,6 +32,9 @@ import org.springframework.data.keyvalue.core.mapping.context.KeyValueMappingCon
 import org.springframework.data.mapping.PersistentProperty;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.model.SimpleTypeHolder;
+import org.springframework.data.redis.core.PartialUpdate;
+import org.springframework.data.redis.core.PartialUpdate.PropertyUpdate;
+import org.springframework.data.redis.core.PartialUpdate.UpdateCommand;
 import org.springframework.data.redis.core.RedisHash;
 import org.springframework.data.redis.core.TimeToLive;
 import org.springframework.data.redis.core.TimeToLiveAccessor;
@@ -42,6 +45,7 @@ import org.springframework.data.redis.core.index.IndexConfiguration;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
+import org.springframework.util.NumberUtils;
 import org.springframework.util.ReflectionUtils;
 import org.springframework.util.ReflectionUtils.MethodCallback;
 import org.springframework.util.ReflectionUtils.MethodFilter;
@@ -75,8 +79,8 @@ public class RedisMappingContext extends KeyValueMappingContext {
 	 */
 	public RedisMappingContext(MappingConfiguration mappingConfiguration) {
 
-		this.mappingConfiguration = mappingConfiguration != null ? mappingConfiguration : new MappingConfiguration(
-				new IndexConfiguration(), new KeyspaceConfiguration());
+		this.mappingConfiguration = mappingConfiguration != null ? mappingConfiguration
+				: new MappingConfiguration(new IndexConfiguration(), new KeyspaceConfiguration());
 
 		setFallbackKeySpaceResolver(new ConfigAwareKeySpaceResolver(this.mappingConfiguration.getKeyspaceConfiguration()));
 		this.timeToLiveAccessor = new ConfigAwareTimeToLiveAccessor(this.mappingConfiguration.getKeyspaceConfiguration(),
@@ -245,7 +249,8 @@ public class RedisMappingContext extends KeyValueMappingContext {
 		public Long getTimeToLive(final Object source) {
 
 			Assert.notNull(source, "Source must not be null!");
-			Class<?> type = source instanceof Class<?> ? (Class<?>) source : source.getClass();
+			Class<?> type = source instanceof Class<?> ? (Class<?>) source
+					: (source instanceof PartialUpdate ? ((PartialUpdate) source).getTarget() : source.getClass());
 
 			Long defaultTimeout = resolveDefaultTimeOut(type);
 			TimeUnit unit = TimeUnit.SECONDS;
@@ -257,12 +262,31 @@ public class RedisMappingContext extends KeyValueMappingContext {
 				if (ttlProperty.findAnnotation(TimeToLive.class) != null) {
 					unit = ttlProperty.findAnnotation(TimeToLive.class).unit();
 				}
+			}
+
+			if (source instanceof PartialUpdate) {
+
+				PartialUpdate<?> update = (PartialUpdate<?>) source;
+
+				if (ttlProperty != null && !update.getPropertyUpdates().isEmpty()) {
+					for (PropertyUpdate pUpdate : update.getPropertyUpdates()) {
+
+						if (UpdateCommand.SET.equals(pUpdate.getCmd()) && ttlProperty.getName().equals(pUpdate.getPropertyPath())) {
+
+							return TimeUnit.SECONDS
+									.convert(NumberUtils.convertNumberToTargetClass((Number) pUpdate.getValue(), Long.class), unit);
+						}
+					}
+				}
+
+			} else if (ttlProperty != null) {
 
 				RedisPersistentEntity entity = mappingContext.getPersistentEntity(type);
 				Long timeout = (Long) entity.getPropertyAccessor(source).getProperty(ttlProperty);
 				if (timeout != null) {
 					return TimeUnit.SECONDS.convert(timeout, unit);
 				}
+
 			} else {
 
 				Method timeoutMethod = resolveTimeMethod(type);
@@ -275,14 +299,14 @@ public class RedisMappingContext extends KeyValueMappingContext {
 							return TimeUnit.SECONDS.convert(timeout.longValue(), ttl.unit());
 						}
 					} catch (IllegalAccessException e) {
-						throw new IllegalStateException("Not allowed to access method '" + timeoutMethod.getName() + "': "
-								+ e.getMessage(), e);
-					} catch (IllegalArgumentException e) {
-						throw new IllegalStateException("Cannot invoke method '" + timeoutMethod.getName()
-								+ " without arguments': " + e.getMessage(), e);
-					} catch (InvocationTargetException e) {
 						throw new IllegalStateException(
-								"Cannot access method '" + timeoutMethod.getName() + "': " + e.getMessage(), e);
+								"Not allowed to access method '" + timeoutMethod.getName() + "': " + e.getMessage(), e);
+					} catch (IllegalArgumentException e) {
+						throw new IllegalStateException(
+								"Cannot invoke method '" + timeoutMethod.getName() + " without arguments': " + e.getMessage(), e);
+					} catch (InvocationTargetException e) {
+						throw new IllegalStateException("Cannot access method '" + timeoutMethod.getName() + "': " + e.getMessage(),
+								e);
 					}
 				}
 			}
diff --git a/src/main/java/org/springframework/data/redis/hash/ObjectHashMapper.java b/src/main/java/org/springframework/data/redis/hash/ObjectHashMapper.java
index fb235512a..96d38ace8 100644
--- a/src/main/java/org/springframework/data/redis/hash/ObjectHashMapper.java
+++ b/src/main/java/org/springframework/data/redis/hash/ObjectHashMapper.java
@@ -143,6 +143,10 @@ public class ObjectHashMapper implements HashMapper<Object, byte[], byte[]> {
 
 		private static final Map<byte[], byte[]> NO_REFERENCE = Collections.emptyMap();
 
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.core.convert.ReferenceResolver#resolveReference(java.io.Serializable, java.lang.String)
+		 */
 		@Override
 		public Map<byte[], byte[]> resolveReference(Serializable id, String keyspace) {
 			return NO_REFERENCE;
@@ -158,9 +162,23 @@ public class ObjectHashMapper implements HashMapper<Object, byte[], byte[]> {
 
 		private static final Set<IndexedData> NO_INDEXES = Collections.emptySet();
 
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.core.convert.IndexResolver#resolveIndexesFor(org.springframework.data.util.TypeInformation, java.lang.Object)
+		 */
 		@Override
 		public Set<IndexedData> resolveIndexesFor(TypeInformation<?> typeInformation, Object value) {
 			return NO_INDEXES;
 		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.core.convert.IndexResolver#resolveIndexesFor(java.lang.String, java.lang.String, org.springframework.data.util.TypeInformation, java.lang.Object)
+		 */
+		@Override
+		public Set<IndexedData> resolveIndexesFor(String keyspace, String path, TypeInformation<?> typeInformation,
+				Object value) {
+			return NO_INDEXES;
+		}
 	}
 }
diff --git a/src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterTests.java b/src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterTests.java
index 4c580c2c2..51451828f 100644
--- a/src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterTests.java
+++ b/src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterTests.java
@@ -21,22 +21,30 @@ import static org.hamcrest.core.IsInstanceOf.*;
 import static org.hamcrest.core.IsNot.*;
 import static org.junit.Assert.*;
 
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.Date;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 
 import org.junit.After;
+import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.Test;
-import org.springframework.beans.factory.DisposableBean;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+import org.springframework.beans.factory.InitializingBean;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.annotation.Reference;
 import org.springframework.data.keyvalue.annotation.KeySpace;
+import org.springframework.data.redis.ConnectionFactoryTracker;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
 import org.springframework.data.redis.core.convert.Bucket;
 import org.springframework.data.redis.core.convert.KeyspaceConfiguration;
 import org.springframework.data.redis.core.convert.MappingConfiguration;
@@ -48,19 +56,35 @@ import org.springframework.data.redis.core.mapping.RedisMappingContext;
  * @author Christoph Strobl
  * @author Mark Paluch
  */
+@RunWith(Parameterized.class)
 public class RedisKeyValueAdapterTests {
 
 	RedisKeyValueAdapter adapter;
 	StringRedisTemplate template;
 	RedisConnectionFactory connectionFactory;
 
+	public RedisKeyValueAdapterTests(RedisConnectionFactory connectionFactory) throws Exception {
+
+		if (connectionFactory instanceof InitializingBean) {
+			((InitializingBean) connectionFactory).afterPropertiesSet();
+		}
+		this.connectionFactory = connectionFactory;
+		ConnectionFactoryTracker.add(connectionFactory);
+	}
+
+	@Parameters
+	public static List<RedisConnectionFactory> params() {
+		return Arrays.<RedisConnectionFactory> asList(new JedisConnectionFactory(), new LettuceConnectionFactory());
+	}
+
+	@AfterClass
+	public static void cleanUp() {
+		ConnectionFactoryTracker.cleanUp();
+	}
+
 	@Before
 	public void setUp() {
 
-		JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory();
-		jedisConnectionFactory.afterPropertiesSet();
-		connectionFactory = jedisConnectionFactory;
-
 		template = new StringRedisTemplate(connectionFactory);
 		template.afterPropertiesSet();
 
@@ -89,14 +113,6 @@ public class RedisKeyValueAdapterTests {
 		} catch (Exception e) {
 			// ignore
 		}
-
-		try {
-			if (connectionFactory instanceof DisposableBean) {
-				((DisposableBean) connectionFactory).destroy();
-			}
-		} catch (Exception e) {
-			// ignore
-		}
 	}
 
 	/**
@@ -312,6 +328,191 @@ public class RedisKeyValueAdapterTests {
 		assertThat(template.opsForSet().isMember("persons:mat:idx", "persons:firstname:mat"), is(true));
 	}
 
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void updateShouldAlterIndexDataCorrectly() {
+
+		Person rand = new Person();
+		rand.firstname = "rand";
+
+		adapter.put("1", rand, "persons");
+
+		assertThat(template.hasKey("persons:firstname:rand"), is(true));
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("1", Person.class) //
+				.set("firstname", "mat");
+
+		adapter.update(update);
+
+		assertThat(template.hasKey("persons:firstname:rand"), is(false));
+		assertThat(template.hasKey("persons:firstname:mat"), is(true));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void updateShouldAlterIndexDataOnNestedObjectCorrectly() {
+
+		Person rand = new Person();
+		rand.address = new Address();
+		rand.address.country = "andor";
+
+		adapter.put("1", rand, "persons");
+
+		assertThat(template.hasKey("persons:address.country:andor"), is(true));
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("1", Person.class);
+		Address addressUpdate = new Address();
+		addressUpdate.country = "tear";
+
+		update = update.set("address", addressUpdate);
+
+		adapter.update(update);
+
+		assertThat(template.hasKey("persons:address.country:andor"), is(false));
+		assertThat(template.hasKey("persons:address.country:tear"), is(true));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void updateShouldAlterIndexDataOnNestedObjectPathCorrectly() {
+
+		Person rand = new Person();
+		rand.address = new Address();
+		rand.address.country = "andor";
+
+		adapter.put("1", rand, "persons");
+
+		assertThat(template.hasKey("persons:address.country:andor"), is(true));
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("1", Person.class) //
+				.set("address.country", "tear");
+
+		adapter.update(update);
+
+		assertThat(template.hasKey("persons:address.country:andor"), is(false));
+		assertThat(template.hasKey("persons:address.country:tear"), is(true));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void updateShouldRemoveComplexObjectCorrectly() {
+
+		Person rand = new Person();
+		rand.address = new Address();
+		rand.address.country = "andor";
+		rand.address.city = "emond's field";
+
+		adapter.put("1", rand, "persons");
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("1", Person.class) //
+				.del("address");
+
+		adapter.update(update);
+
+		assertThat(template.opsForHash().hasKey("persons:1", "address.country"), is(false));
+		assertThat(template.opsForHash().hasKey("persons:1", "address.city"), is(false));
+		assertThat(template.opsForSet().isMember("persons:address.country:andor", "1"), is(false));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void updateShouldRemoveSimpleListValuesCorrectly() {
+
+		Person rand = new Person();
+		rand.nicknames = Arrays.asList("lews therin", "dragon reborn");
+
+		adapter.put("1", rand, "persons");
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("1", Person.class) //
+				.del("nicknames");
+
+		adapter.update(update);
+
+		assertThat(template.opsForHash().hasKey("persons:1", "nicknames.[0]"), is(false));
+		assertThat(template.opsForHash().hasKey("persons:1", "nicknames.[1]"), is(false));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void updateShouldRemoveComplexListValuesCorrectly() {
+
+		Person mat = new Person();
+		mat.firstname = "mat";
+		mat.nicknames = Collections.singletonList("prince of ravens");
+
+		Person perrin = new Person();
+		perrin.firstname = "mat";
+		perrin.nicknames = Collections.singletonList("lord of the two rivers");
+
+		Person rand = new Person();
+		rand.coworkers = Arrays.asList(mat, perrin);
+
+		adapter.put("1", rand, "persons");
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("1", Person.class) //
+				.del("coworkers");
+
+		adapter.update(update);
+
+		assertThat(template.opsForHash().hasKey("persons:1", "coworkers.[0].firstname"), is(false));
+		assertThat(template.opsForHash().hasKey("persons:1", "coworkers.[0].nicknames.[0]"), is(false));
+		assertThat(template.opsForHash().hasKey("persons:1", "coworkers.[1].firstname"), is(false));
+		assertThat(template.opsForHash().hasKey("persons:1", "coworkers.[1].nicknames.[0]"), is(false));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void updateShouldRemoveSimpleMapValuesCorrectly() {
+
+		Person rand = new Person();
+		rand.physicalAttributes = Collections.singletonMap("eye-color", "grey");
+
+		adapter.put("1", rand, "persons");
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("1", Person.class) //
+				.del("physicalAttributes");
+
+		adapter.update(update);
+
+		assertThat(template.opsForHash().hasKey("persons:1", "physicalAttributes.[eye-color]"), is(false));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void updateShouldRemoveComplexMapValuesCorrectly() {
+
+		Person tam = new Person();
+		tam.firstname = "tam";
+
+		Person rand = new Person();
+		rand.relatives = Collections.singletonMap("stepfather", tam);
+
+		adapter.put("1", rand, "persons");
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("1", Person.class) //
+				.del("relatives");
+
+		adapter.update(update);
+
+		assertThat(template.opsForHash().hasKey("persons:1", "relatives.[stepfather].firstname"), is(false));
+	}
+
 	@KeySpace("persons")
 	static class Person {
 
diff --git a/src/test/java/org/springframework/data/redis/core/RedisKeyValueTemplateTests.java b/src/test/java/org/springframework/data/redis/core/RedisKeyValueTemplateTests.java
index 88d69c567..fd57a3987 100644
--- a/src/test/java/org/springframework/data/redis/core/RedisKeyValueTemplateTests.java
+++ b/src/test/java/org/springframework/data/redis/core/RedisKeyValueTemplateTests.java
@@ -17,11 +17,15 @@ package org.springframework.data.redis.core;
 
 import static org.hamcrest.core.Is.*;
 import static org.hamcrest.core.IsCollectionContaining.*;
+import static org.hamcrest.core.IsEqual.*;
 import static org.junit.Assert.*;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 
 import org.junit.After;
 import org.junit.AfterClass;
@@ -36,9 +40,13 @@ import org.springframework.data.redis.ConnectionFactoryTracker;
 import org.springframework.data.redis.connection.RedisConnection;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.core.index.Indexed;
 import org.springframework.data.redis.core.mapping.RedisMappingContext;
 import org.springframework.util.ObjectUtils;
 
+import lombok.EqualsAndHashCode;
+
 /**
  * @author Christoph Strobl
  */
@@ -48,11 +56,14 @@ public class RedisKeyValueTemplateTests {
 	RedisConnectionFactory connectionFactory;
 	RedisKeyValueTemplate template;
 	RedisTemplate<Object, Object> nativeTemplate;
+	RedisMappingContext context;
+	RedisKeyValueAdapter adapter;
 
 	public RedisKeyValueTemplateTests(RedisConnectionFactory connectionFactory) {
 
 		this.connectionFactory = connectionFactory;
 		ConnectionFactoryTracker.add(connectionFactory);
+
 	}
 
 	@Parameters
@@ -61,7 +72,10 @@ public class RedisKeyValueTemplateTests {
 		JedisConnectionFactory jedis = new JedisConnectionFactory();
 		jedis.afterPropertiesSet();
 
-		return Collections.<RedisConnectionFactory> singletonList(jedis);
+		LettuceConnectionFactory lettuce = new LettuceConnectionFactory();
+		lettuce.afterPropertiesSet();
+
+		return Arrays.<RedisConnectionFactory> asList(jedis, lettuce);
 	}
 
 	@AfterClass
@@ -76,14 +90,13 @@ public class RedisKeyValueTemplateTests {
 		nativeTemplate.setConnectionFactory(connectionFactory);
 		nativeTemplate.afterPropertiesSet();
 
-		RedisMappingContext context = new RedisMappingContext();
-
-		RedisKeyValueAdapter adapter = new RedisKeyValueAdapter(nativeTemplate, context);
+		context = new RedisMappingContext();
+		adapter = new RedisKeyValueAdapter(nativeTemplate, context);
 		template = new RedisKeyValueTemplate(adapter, context);
 	}
 
 	@After
-	public void tearDown() {
+	public void tearDown() throws Exception {
 
 		nativeTemplate.execute(new RedisCallback<Void>() {
 
@@ -94,6 +107,8 @@ public class RedisKeyValueTemplateTests {
 				return null;
 			}
 		});
+
+		template.destroy();
 	}
 
 	/**
@@ -198,16 +213,650 @@ public class RedisKeyValueTemplateTests {
 		assertThat(result.size(), is(0));
 	}
 
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void partialUpdate() {
+
+		final Person rand = new Person();
+		rand.firstname = "rand";
+
+		template.insert(rand);
+
+		/*
+		 * Set the lastname and make sure we've an index on it afterwards
+		 */
+		Person update1 = new Person(rand.id, null, "al-thor");
+		PartialUpdate<Person> update = new PartialUpdate<Person>(rand.id, update1);
+
+		template.doPartialUpdate(update);
+
+		assertThat(template.findById(rand.id, Person.class), is(equalTo(new Person(rand.id, "rand", "al-thor"))));
+		nativeTemplate.execute(new RedisCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisConnection connection) throws DataAccessException {
+
+				assertThat(connection.hGet(("template-test-person:" + rand.id).getBytes(), "firstname".getBytes()),
+						is(equalTo("rand".getBytes())));
+				assertThat(connection.exists("template-test-person:lastname:al-thor".getBytes()), is(true));
+				assertThat(connection.sIsMember("template-test-person:lastname:al-thor".getBytes(), rand.id.getBytes()),
+						is(true));
+				return null;
+			}
+		});
+
+		/*
+		 * Set the firstname and make sure lastname index and value is not affected
+		 */
+		update = new PartialUpdate<Person>(rand.id, Person.class).set("firstname", "frodo");
+
+		template.doPartialUpdate(update);
+
+		assertThat(template.findById(rand.id, Person.class), is(equalTo(new Person(rand.id, "frodo", "al-thor"))));
+		nativeTemplate.execute(new RedisCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisConnection connection) throws DataAccessException {
+
+				assertThat(connection.exists("template-test-person:lastname:al-thor".getBytes()), is(true));
+				assertThat(connection.sIsMember("template-test-person:lastname:al-thor".getBytes(), rand.id.getBytes()),
+						is(true));
+				return null;
+			}
+		});
+
+		/*
+		 * Remote firstname and update lastname. Make sure lastname index is updated
+		 */
+		update = new PartialUpdate<Person>(rand.id, Person.class) //
+				.del("firstname").set("lastname", "baggins");
+
+		template.doPartialUpdate(update);
+
+		assertThat(template.findById(rand.id, Person.class), is(equalTo(new Person(rand.id, null, "baggins"))));
+		nativeTemplate.execute(new RedisCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisConnection connection) throws DataAccessException {
+
+				assertThat(connection.exists("template-test-person:lastname:al-thor".getBytes()), is(false));
+				assertThat(connection.exists("template-test-person:lastname:baggins".getBytes()), is(true));
+				assertThat(connection.sIsMember("template-test-person:lastname:baggins".getBytes(), rand.id.getBytes()),
+						is(true));
+				return null;
+			}
+		});
+
+		/*
+		 * Remove lastname and make sure the index vanishes
+		 */
+		update = new PartialUpdate<Person>(rand.id, Person.class) //
+				.del("lastname");
+
+		template.doPartialUpdate(update);
+
+		assertThat(template.findById(rand.id, Person.class), is(equalTo(new Person(rand.id, null, null))));
+		nativeTemplate.execute(new RedisCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisConnection connection) throws DataAccessException {
+
+				assertThat(connection.keys("template-test-person:lastname:*".getBytes()).size(), is(0));
+				return null;
+			}
+		});
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void partialUpdateSimpleType() {
+
+		final VariousTypes source = new VariousTypes();
+		source.stringValue = "some-value";
+
+		template.insert(source);
+
+		PartialUpdate<VariousTypes> update = new PartialUpdate<VariousTypes>(source.id, VariousTypes.class) //
+				.set("stringValue", "hooya!");
+
+		template.doPartialUpdate(update);
+
+		nativeTemplate.execute(new RedisCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisConnection connection) throws DataAccessException {
+
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(), "stringValue".getBytes()),
+						is("hooya!".getBytes()));
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"simpleTypedMap._class".getBytes()), is(false));
+				return null;
+			}
+		});
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void partialUpdateComplexType() {
+
+		Item callandor = new Item();
+		callandor.name = "Callandor";
+		callandor.dimension = new Dimension();
+		callandor.dimension.length = 100;
+
+		final VariousTypes source = new VariousTypes();
+		source.complexValue = callandor;
+
+		template.insert(source);
+
+		Item portalStone = new Item();
+		portalStone.name = "Portal Stone";
+		portalStone.dimension = new Dimension();
+		portalStone.dimension.height = 350;
+		portalStone.dimension.width = 70;
+
+		PartialUpdate<VariousTypes> update = new PartialUpdate<VariousTypes>(source.id, VariousTypes.class) //
+				.set("complexValue", portalStone);
+
+		template.doPartialUpdate(update);
+
+		nativeTemplate.execute(new RedisCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisConnection connection) throws DataAccessException {
+
+				assertThat(
+						connection.hGet(("template-test-type-mapping:" + source.id).getBytes(), "complexValue.name".getBytes()),
+						is("Portal Stone".getBytes()));
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"complexValue.dimension.height".getBytes()), is("350".getBytes()));
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"complexValue.dimension.width".getBytes()), is("70".getBytes()));
+
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"complexValue.dimension.length".getBytes()), is(false));
+				return null;
+			}
+		});
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void partialUpdateObjectType() {
+
+		Item callandor = new Item();
+		callandor.name = "Callandor";
+		callandor.dimension = new Dimension();
+		callandor.dimension.length = 100;
+
+		final VariousTypes source = new VariousTypes();
+		source.objectValue = callandor;
+
+		template.insert(source);
+
+		Item portalStone = new Item();
+		portalStone.name = "Portal Stone";
+		portalStone.dimension = new Dimension();
+		portalStone.dimension.height = 350;
+		portalStone.dimension.width = 70;
+
+		PartialUpdate<VariousTypes> update = new PartialUpdate<VariousTypes>(source.id, VariousTypes.class) //
+				.set("objectValue", portalStone);
+
+		template.doPartialUpdate(update);
+
+		nativeTemplate.execute(new RedisCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisConnection connection) throws DataAccessException {
+
+				assertThat(
+						connection.hGet(("template-test-type-mapping:" + source.id).getBytes(), "objectValue._class".getBytes()),
+						is(Item.class.getName().getBytes()));
+				assertThat(
+						connection.hGet(("template-test-type-mapping:" + source.id).getBytes(), "objectValue.name".getBytes()),
+						is("Portal Stone".getBytes()));
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"objectValue.dimension.height".getBytes()), is("350".getBytes()));
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"objectValue.dimension.width".getBytes()), is("70".getBytes()));
+
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(), "objectValue".getBytes()),
+						is(false));
+				return null;
+			}
+		});
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void partialUpdateSimpleTypedMap() {
+
+		final VariousTypes source = new VariousTypes();
+		source.simpleTypedMap = new LinkedHashMap<String, String>();
+		source.simpleTypedMap.put("key-1", "rand");
+		source.simpleTypedMap.put("key-2", "mat");
+		source.simpleTypedMap.put("key-3", "perrin");
+
+		template.insert(source);
+
+		PartialUpdate<VariousTypes> update = new PartialUpdate<VariousTypes>(source.id, VariousTypes.class) //
+				.set("simpleTypedMap", Collections.singletonMap("spring", "data"));
+
+		template.doPartialUpdate(update);
+
+		nativeTemplate.execute(new RedisCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisConnection connection) throws DataAccessException {
+
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"simpleTypedMap.[spring]".getBytes()), is("data".getBytes()));
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"simpleTypedMap.[key-1]".getBytes()), is(false));
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"simpleTypedMap.[key-2]".getBytes()), is(false));
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"simpleTypedMap.[key-2]".getBytes()), is(false));
+				return null;
+			}
+		});
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void partialUpdateComplexTypedMap() {
+
+		final VariousTypes source = new VariousTypes();
+		source.complexTypedMap = new LinkedHashMap<String, Item>();
+
+		Item callandor = new Item();
+		callandor.name = "Callandor";
+		callandor.dimension = new Dimension();
+		callandor.dimension.height = 100;
+
+		Item portalStone = new Item();
+		portalStone.name = "Portal Stone";
+		portalStone.dimension = new Dimension();
+		portalStone.dimension.height = 350;
+		portalStone.dimension.width = 70;
+
+		source.complexTypedMap.put("callandor", callandor);
+		source.complexTypedMap.put("portal-stone", portalStone);
+
+		template.insert(source);
+
+		Item hornOfValere = new Item();
+		hornOfValere.name = "Horn of Valere";
+		hornOfValere.dimension = new Dimension();
+		hornOfValere.dimension.height = 70;
+		hornOfValere.dimension.width = 25;
+
+		PartialUpdate<VariousTypes> update = new PartialUpdate<VariousTypes>(source.id, VariousTypes.class) //
+				.set("complexTypedMap", Collections.singletonMap("horn-of-valere", hornOfValere));
+
+		template.doPartialUpdate(update);
+
+		nativeTemplate.execute(new RedisCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisConnection connection) throws DataAccessException {
+
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"complexTypedMap.[horn-of-valere].name".getBytes()), is("Horn of Valere".getBytes()));
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"complexTypedMap.[horn-of-valere].dimension.height".getBytes()), is("70".getBytes()));
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"complexTypedMap.[horn-of-valere].dimension.width".getBytes()), is("25".getBytes()));
+
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"complexTypedMap.[callandor].name".getBytes()), is(false));
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"complexTypedMap.[callandor].dimension.height".getBytes()), is(false));
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"complexTypedMap.[callandor].dimension.width".getBytes()), is(false));
+
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"complexTypedMap.[portal-stone].name".getBytes()), is(false));
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"complexTypedMap.[portal-stone].dimension.height".getBytes()), is(false));
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"complexTypedMap.[portal-stone].dimension.width".getBytes()), is(false));
+				return null;
+			}
+		});
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void partialUpdateObjectTypedMap() {
+
+		final VariousTypes source = new VariousTypes();
+		source.untypedMap = new LinkedHashMap<String, Object>();
+
+		Item callandor = new Item();
+		callandor.name = "Callandor";
+		callandor.dimension = new Dimension();
+		callandor.dimension.height = 100;
+
+		Item portalStone = new Item();
+		portalStone.name = "Portal Stone";
+		portalStone.dimension = new Dimension();
+		portalStone.dimension.height = 350;
+		portalStone.dimension.width = 70;
+
+		source.untypedMap.put("callandor", callandor);
+		source.untypedMap.put("just-a-string", "some-string-value");
+		source.untypedMap.put("portal-stone", portalStone);
+
+		template.insert(source);
+
+		Item hornOfValere = new Item();
+		hornOfValere.name = "Horn of Valere";
+		hornOfValere.dimension = new Dimension();
+		hornOfValere.dimension.height = 70;
+		hornOfValere.dimension.width = 25;
+
+		Map<String, Object> map = new LinkedHashMap<String, Object>();
+		map.put("spring", "data");
+		map.put("horn-of-valere", hornOfValere);
+		map.put("some-number", 100L);
+
+		PartialUpdate<VariousTypes> update = new PartialUpdate<VariousTypes>(source.id, VariousTypes.class) //
+				.set("untypedMap", map);
+
+		template.doPartialUpdate(update);
+
+		nativeTemplate.execute(new RedisCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisConnection connection) throws DataAccessException {
+
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"untypedMap.[horn-of-valere].name".getBytes()), is("Horn of Valere".getBytes()));
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"untypedMap.[horn-of-valere].dimension.height".getBytes()), is("70".getBytes()));
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"untypedMap.[horn-of-valere].dimension.width".getBytes()), is("25".getBytes()));
+
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"untypedMap.[spring]._class".getBytes()), is("java.lang.String".getBytes()));
+				assertThat(
+						connection.hGet(("template-test-type-mapping:" + source.id).getBytes(), "untypedMap.[spring]".getBytes()),
+						is("data".getBytes()));
+
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"untypedMap.[some-number]._class".getBytes()), is("java.lang.Long".getBytes()));
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"untypedMap.[some-number]".getBytes()), is("100".getBytes()));
+
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"untypedMap.[callandor].name".getBytes()), is(false));
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"untypedMap.[callandor].dimension.height".getBytes()), is(false));
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"untypedMap.[callandor].dimension.width".getBytes()), is(false));
+
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"untypedMap.[portal-stone].name".getBytes()), is(false));
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"untypedMap.[portal-stone].dimension.height".getBytes()), is(false));
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"untypedMap.[portal-stone].dimension.width".getBytes()), is(false));
+				return null;
+			}
+		});
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void partialUpdateSimpleTypedList() {
+
+		final VariousTypes source = new VariousTypes();
+		source.simpleTypedList = new ArrayList<String>();
+		source.simpleTypedList.add("rand");
+		source.simpleTypedList.add("mat");
+		source.simpleTypedList.add("perrin");
+
+		template.insert(source);
+
+		PartialUpdate<VariousTypes> update = new PartialUpdate<VariousTypes>(source.id, VariousTypes.class) //
+				.set("simpleTypedList", Collections.singletonList("spring"));
+
+		template.doPartialUpdate(update);
+
+		nativeTemplate.execute(new RedisCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisConnection connection) throws DataAccessException {
+
+				assertThat(
+						connection.hGet(("template-test-type-mapping:" + source.id).getBytes(), "simpleTypedList.[0]".getBytes()),
+						is("spring".getBytes()));
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"simpleTypedList.[1]".getBytes()), is(false));
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"simpleTypedList.[2]".getBytes()), is(false));
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"simpleTypedList.[3]".getBytes()), is(false));
+				return null;
+			}
+		});
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void partialUpdateComplexTypedList() {
+
+		final VariousTypes source = new VariousTypes();
+		source.complexTypedList = new ArrayList<Item>();
+
+		Item callandor = new Item();
+		callandor.name = "Callandor";
+		callandor.dimension = new Dimension();
+		callandor.dimension.height = 100;
+
+		Item portalStone = new Item();
+		portalStone.name = "Portal Stone";
+		portalStone.dimension = new Dimension();
+		portalStone.dimension.height = 350;
+		portalStone.dimension.width = 70;
+
+		source.complexTypedList.add(callandor);
+		source.complexTypedList.add(portalStone);
+
+		template.insert(source);
+
+		Item hornOfValere = new Item();
+		hornOfValere.name = "Horn of Valere";
+		hornOfValere.dimension = new Dimension();
+		hornOfValere.dimension.height = 70;
+		hornOfValere.dimension.width = 25;
+
+		PartialUpdate<VariousTypes> update = new PartialUpdate<VariousTypes>(source.id, VariousTypes.class) //
+				.set("complexTypedList", Collections.singletonList(hornOfValere));
+
+		template.doPartialUpdate(update);
+
+		nativeTemplate.execute(new RedisCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisConnection connection) throws DataAccessException {
+
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"complexTypedList.[0].name".getBytes()), is("Horn of Valere".getBytes()));
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"complexTypedList.[0].dimension.height".getBytes()), is("70".getBytes()));
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"complexTypedList.[0].dimension.width".getBytes()), is("25".getBytes()));
+
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"complexTypedMap.[1].name".getBytes()), is(false));
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"complexTypedMap.[1].dimension.height".getBytes()), is(false));
+				assertThat(connection.hExists(("template-test-type-mapping:" + source.id).getBytes(),
+						"complexTypedMap.[1].dimension.width".getBytes()), is(false));
+				return null;
+			}
+		});
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void partialUpdateObjectTypedList() {
+
+		final VariousTypes source = new VariousTypes();
+		source.untypedList = new ArrayList<Object>();
+
+		Item callandor = new Item();
+		callandor.name = "Callandor";
+		callandor.dimension = new Dimension();
+		callandor.dimension.height = 100;
+
+		Item portalStone = new Item();
+		portalStone.name = "Portal Stone";
+		portalStone.dimension = new Dimension();
+		portalStone.dimension.height = 350;
+		portalStone.dimension.width = 70;
+
+		source.untypedList.add(callandor);
+		source.untypedList.add("some-string-value");
+		source.untypedList.add(portalStone);
+
+		template.insert(source);
+
+		Item hornOfValere = new Item();
+		hornOfValere.name = "Horn of Valere";
+		hornOfValere.dimension = new Dimension();
+		hornOfValere.dimension.height = 70;
+		hornOfValere.dimension.width = 25;
+
+		List<Object> list = new ArrayList<Object>();
+		list.add("spring");
+		list.add(hornOfValere);
+		list.add(100L);
+
+		PartialUpdate<VariousTypes> update = new PartialUpdate<VariousTypes>(source.id, VariousTypes.class) //
+				.set("untypedList", list);
+
+		template.doPartialUpdate(update);
+
+		nativeTemplate.execute(new RedisCallback<Void>() {
+
+			@Override
+			public Void doInRedis(RedisConnection connection) throws DataAccessException {
+
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"untypedList.[0]._class".getBytes()), is("java.lang.String".getBytes()));
+				assertThat(
+						connection.hGet(("template-test-type-mapping:" + source.id).getBytes(), "untypedList.[0]".getBytes()),
+						is("spring".getBytes()));
+
+				assertThat(
+						connection.hGet(("template-test-type-mapping:" + source.id).getBytes(), "untypedList.[1].name".getBytes()),
+						is("Horn of Valere".getBytes()));
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"untypedList.[1].dimension.height".getBytes()), is("70".getBytes()));
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"untypedList.[1].dimension.width".getBytes()), is("25".getBytes()));
+
+				assertThat(connection.hGet(("template-test-type-mapping:" + source.id).getBytes(),
+						"untypedList.[2]._class".getBytes()), is("java.lang.Long".getBytes()));
+				assertThat(
+						connection.hGet(("template-test-type-mapping:" + source.id).getBytes(), "untypedList.[2]".getBytes()),
+						is("100".getBytes()));
+
+				return null;
+			}
+		});
+	}
+
+	@EqualsAndHashCode
+	@RedisHash("template-test-type-mapping")
+	static class VariousTypes {
+
+		@Id String id;
+
+		String stringValue;
+		Integer integerValue;
+		Item complexValue;
+		Object objectValue;
+
+		List<String> simpleTypedList;
+		List<Item> complexTypedList;
+		List<Object> untypedList;
+
+		Map<String, String> simpleTypedMap;
+		Map<String, Item> complexTypedMap;
+		Map<String, Object> untypedMap;
+	}
+
+	static class Item {
+		String name;
+		Dimension dimension;
+	}
+
+	static class Dimension {
+
+		Integer height;
+		Integer width;
+		Integer length;
+	}
+
 	@RedisHash("template-test-person")
 	static class Person {
 
 		@Id String id;
 		String firstname;
+		@Indexed String lastname;
+		Integer age;
+		List<String> nicknames;
+
+		public Person() {}
+
+		public Person(String firstname, String lastname) {
+			this(null, firstname, lastname, null);
+		}
+
+		public Person(String id, String firstname, String lastname) {
+			this(id, firstname, lastname, null);
+		}
+
+		public Person(String id, String firstname, String lastname, Integer age) {
+
+			this.id = id;
+			this.firstname = firstname;
+			this.lastname = lastname;
+			this.age = age;
+		}
 
 		@Override
 		public int hashCode() {
 
 			int result = ObjectUtils.nullSafeHashCode(firstname);
+			result += ObjectUtils.nullSafeHashCode(lastname);
+			result += ObjectUtils.nullSafeHashCode(age);
+			result += ObjectUtils.nullSafeHashCode(nicknames);
 			return result + ObjectUtils.nullSafeHashCode(id);
 		}
 
@@ -228,8 +877,25 @@ public class RedisKeyValueTemplateTests {
 				return false;
 			}
 
+			if (!ObjectUtils.nullSafeEquals(this.lastname, that.lastname)) {
+				return false;
+			}
+
+			if (!ObjectUtils.nullSafeEquals(this.age, that.age)) {
+				return false;
+			}
+			if (!ObjectUtils.nullSafeEquals(this.nicknames, that.nicknames)) {
+				return false;
+			}
+
 			return ObjectUtils.nullSafeEquals(this.id, that.id);
 		}
 
+		@Override
+		public String toString() {
+			return "Person [id=" + id + ", firstname=" + firstname + ", lastname=" + lastname + ", age=" + age
+					+ ", nicknames=" + nicknames + "]";
+		}
+
 	}
 }
diff --git a/src/test/java/org/springframework/data/redis/core/convert/MappingRedisConverterUnitTests.java b/src/test/java/org/springframework/data/redis/core/convert/MappingRedisConverterUnitTests.java
index 867c58a14..ac2302fbc 100644
--- a/src/test/java/org/springframework/data/redis/core/convert/MappingRedisConverterUnitTests.java
+++ b/src/test/java/org/springframework/data/redis/core/convert/MappingRedisConverterUnitTests.java
@@ -49,13 +49,17 @@ import java.util.UUID;
 
 import org.hamcrest.core.IsEqual;
 import org.junit.Before;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.ExpectedException;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.runners.MockitoJUnitRunner;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.data.convert.ReadingConverter;
 import org.springframework.data.convert.WritingConverter;
+import org.springframework.data.mapping.model.MappingException;
+import org.springframework.data.redis.core.PartialUpdate;
 import org.springframework.data.redis.core.convert.ConversionTestEntities.Address;
 import org.springframework.data.redis.core.convert.ConversionTestEntities.AddressWithId;
 import org.springframework.data.redis.core.convert.ConversionTestEntities.AddressWithPostcode;
@@ -84,6 +88,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 @RunWith(MockitoJUnitRunner.class)
 public class MappingRedisConverterUnitTests {
 
+	public @Rule ExpectedException exception = ExpectedException.none();
 	@Mock ReferenceResolver resolverMock;
 	MappingRedisConverter converter;
 	Person rand;
@@ -95,7 +100,6 @@ public class MappingRedisConverterUnitTests {
 		converter.afterPropertiesSet();
 
 		rand = new Person();
-
 	}
 
 	/**
@@ -1582,6 +1586,8 @@ public class MappingRedisConverterUnitTests {
 	}
 
 	/**
+	 * <<<<<<< HEAD
+	 * 
 	 * @see DATAREDIS-509
 	 */
 	@Test
@@ -1611,6 +1617,405 @@ public class MappingRedisConverterUnitTests {
 				.containingUtf8String("arrayOfPrimitives.[1]", "2").containingUtf8String("arrayOfPrimitives.[2]", "3"));
 	}
 
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldNotAppendClassTypeHint() {
+
+		Person value = new Person();
+		value.firstname = "rand";
+		value.age = 24;
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", value);
+
+		assertThat(write(update).getBucket().get("_class"), is(nullValue()));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdateSimpleValueCorrectly() {
+
+		Person value = new Person();
+		value.firstname = "rand";
+		value.age = 24;
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", value);
+
+		assertThat(write(update).getBucket(),
+				isBucket().containingUtf8String("firstname", "rand").containingUtf8String("age", "24"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdatePathWithSimpleValueCorrectly() {
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class).set("firstname", "rand").set("age",
+				24);
+
+		assertThat(write(update).getBucket(),
+				isBucket().containingUtf8String("firstname", "rand").containingUtf8String("age", "24"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdateNestedPathWithSimpleValueCorrectly() {
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class).set("address.city", "two rivers");
+
+		assertThat(write(update).getBucket(), isBucket().containingUtf8String("address.city", "two rivers"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdatePathWithComplexValueCorrectly() {
+
+		Address address = new Address();
+		address.city = "two rivers";
+		address.country = "andor";
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class).set("address", address);
+
+		assertThat(write(update).getBucket(),
+				isBucket().containingUtf8String("address.city", "two rivers").containingUtf8String("address.country", "andor"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdatePathWithSimpleListValueCorrectly() {
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class).set("nicknames",
+				Arrays.asList("dragon", "lews"));
+
+		assertThat(write(update).getBucket(),
+				isBucket().containingUtf8String("nicknames.[0]", "dragon").containingUtf8String("nicknames.[1]", "lews"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdatePathWithComplexListValueCorrectly() {
+
+		Person mat = new Person();
+		mat.firstname = "mat";
+		mat.age = 24;
+
+		Person perrin = new Person();
+		perrin.firstname = "perrin";
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class).set("coworkers",
+				Arrays.asList(mat, perrin));
+
+		assertThat(write(update).getBucket(), isBucket().containingUtf8String("coworkers.[0].firstname", "mat")
+				.containingUtf8String("coworkers.[0].age", "24").containingUtf8String("coworkers.[1].firstname", "perrin"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdatePathWithSimpleListValueWhenNotPassedInAsCollectionCorrectly() {
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class).set("nicknames", "dragon");
+
+		assertThat(write(update).getBucket(), isBucket().containingUtf8String("nicknames.[0]", "dragon"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdatePathWithComplexListValueWhenNotPassedInAsCollectionCorrectly() {
+
+		Person mat = new Person();
+		mat.firstname = "mat";
+		mat.age = 24;
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class).set("coworkers", mat);
+
+		assertThat(write(update).getBucket(), isBucket().containingUtf8String("coworkers.[0].firstname", "mat")
+				.containingUtf8String("coworkers.[0].age", "24"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdatePathWithSimpleListValueWhenNotPassedInAsCollectionWithPositionalParameterCorrectly() {
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class).set("nicknames.[5]", "dragon");
+
+		assertThat(write(update).getBucket(), isBucket().containingUtf8String("nicknames.[5]", "dragon"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdatePathWithComplexListValueWhenNotPassedInAsCollectionWithPositionalParameterCorrectly() {
+
+		Person mat = new Person();
+		mat.firstname = "mat";
+		mat.age = 24;
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class).set("coworkers.[5]", mat);
+
+		assertThat(write(update).getBucket(), isBucket().containingUtf8String("coworkers.[5].firstname", "mat")
+				.containingUtf8String("coworkers.[5].age", "24"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdatePathWithSimpleMapValueCorrectly() {
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class).set("physicalAttributes",
+				Collections.singletonMap("eye-color", "grey"));
+
+		assertThat(write(update).getBucket(), isBucket().containingUtf8String("physicalAttributes.[eye-color]", "grey"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdatePathWithComplexMapValueCorrectly() {
+
+		Person tam = new Person();
+		tam.firstname = "tam";
+		tam.alive = false;
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class).set("relatives",
+				Collections.singletonMap("father", tam));
+
+		assertThat(write(update).getBucket(), isBucket().containingUtf8String("relatives.[father].firstname", "tam")
+				.containingUtf8String("relatives.[father].alive", "0"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdatePathWithSimpleMapValueWhenNotPassedInAsCollectionCorrectly() {
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class).set("physicalAttributes",
+				Collections.singletonMap("eye-color", "grey").entrySet().iterator().next());
+
+		assertThat(write(update).getBucket(), isBucket().containingUtf8String("physicalAttributes.[eye-color]", "grey"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdatePathWithComplexMapValueWhenNotPassedInAsCollectionCorrectly() {
+
+		Person tam = new Person();
+		tam.firstname = "tam";
+		tam.alive = false;
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class).set("relatives",
+				Collections.singletonMap("father", tam).entrySet().iterator().next());
+
+		assertThat(write(update).getBucket(), isBucket().containingUtf8String("relatives.[father].firstname", "tam")
+				.containingUtf8String("relatives.[father].alive", "0"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdatePathWithSimpleMapValueWhenNotPassedInAsCollectionWithPositionalParameterCorrectly() {
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class).set("physicalAttributes.[eye-color]",
+				"grey");
+
+		assertThat(write(update).getBucket(), isBucket().containingUtf8String("physicalAttributes.[eye-color]", "grey"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdatePathWithSimpleMapValueOnNestedElementCorrectly() {
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class).set("relatives.[father].firstname",
+				"tam");
+
+		assertThat(write(update).getBucket(), isBucket().containingUtf8String("relatives.[father].firstname", "tam"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test(expected = MappingException.class)
+	public void writeShouldThrowExceptionOnPartialUpdatePathWithSimpleMapValueWhenItsASingleValueWithoutPath() {
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class).set("physicalAttributes", "grey");
+
+		write(update);
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdatePathWithRegisteredCustomConversionCorrectly() {
+
+		this.converter = new MappingRedisConverter(null, null, resolverMock);
+		this.converter
+				.setCustomConversions(new CustomConversions(Collections.singletonList(new AddressToBytesConverter())));
+		this.converter.afterPropertiesSet();
+
+		Address address = new Address();
+		address.country = "Tel'aran'rhiod";
+		address.city = "unknown";
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class).set("address", address);
+
+		assertThat(write(update).getBucket(),
+				isBucket().containingUtf8String("address", "{\"city\":\"unknown\",\"country\":\"Tel'aran'rhiod\"}"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdatePathWithReferenceCorrectly() {
+
+		Location tar = new Location();
+		tar.id = "1";
+		tar.name = "tar valon";
+
+		Location tear = new Location();
+		tear.id = "2";
+		tear.name = "city of tear";
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class).set("visited",
+				Arrays.asList(tar, tear));
+
+		assertThat(write(update).getBucket(),
+				isBucket().containingUtf8String("visited.[0]", "locations:1").containingUtf8String("visited.[1]", "locations:2") //
+						.without("visited.id") //
+						.without("visited.name"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldWritePartialUpdatePathWithListOfReferencesCorrectly() {
+
+		Location location = new Location();
+		location.id = "1";
+		location.name = "tar valon";
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class) //
+				.set("location", location);
+
+		assertThat(write(update).getBucket(),
+				isBucket().containingUtf8String("location", "locations:1") //
+						.without("location.id") //
+						.without("location.name"));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldThrowExceptionForUpdateValueNotAssignableToDomainTypeProperty() {
+
+		exception.expect(MappingException.class);
+		exception.expectMessage("java.lang.String cannot be assigned");
+		exception.expectMessage("java.lang.Integer");
+		exception.expectMessage("age");
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class) //
+				.set("age", "twenty-four");
+
+		assertThat(write(update).getBucket().get("_class"), is(nullValue()));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldThrowExceptionForUpdateCollectionValueNotAssignableToDomainTypeProperty() {
+
+		exception.expect(MappingException.class);
+		exception.expectMessage("java.lang.String cannot be assigned");
+		exception.expectMessage(Person.class.getName());
+		exception.expectMessage("coworkers.[0]");
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class) //
+				.set("coworkers.[0]", "buh buh the bear");
+
+		assertThat(write(update).getBucket().get("_class"), is(nullValue()));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldThrowExceptionForUpdateValueInCollectionNotAssignableToDomainTypeProperty() {
+
+		exception.expect(MappingException.class);
+		exception.expectMessage("java.lang.String cannot be assigned");
+		exception.expectMessage(Person.class.getName());
+		exception.expectMessage("coworkers");
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class) //
+				.set("coworkers", Collections.singletonList("foo"));
+
+		assertThat(write(update).getBucket().get("_class"), is(nullValue()));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldThrowExceptionForUpdateMapValueNotAssignableToDomainTypeProperty() {
+
+		exception.expect(MappingException.class);
+		exception.expectMessage("java.lang.String cannot be assigned");
+		exception.expectMessage(Person.class.getName());
+		exception.expectMessage("relatives.[father]");
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class) //
+				.set("relatives.[father]", "buh buh the bear");
+
+		assertThat(write(update).getBucket().get("_class"), is(nullValue()));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void writeShouldThrowExceptionForUpdateValueInMapNotAssignableToDomainTypeProperty() {
+
+		exception.expect(MappingException.class);
+		exception.expectMessage("java.lang.String cannot be assigned");
+		exception.expectMessage(Person.class.getName());
+		exception.expectMessage("relatives.[father]");
+
+		PartialUpdate<Person> update = new PartialUpdate<Person>("123", Person.class) //
+				.set("relatives", Collections.singletonMap("father", "buh buh the bear"));
+
+		assertThat(write(update).getBucket().get("_class"), is(nullValue()));
+	}
+
 	private RedisData write(Object source) {
 
 		RedisData rdo = new RedisData();
diff --git a/src/test/java/org/springframework/data/redis/core/convert/PathIndexResolverUnitTests.java b/src/test/java/org/springframework/data/redis/core/convert/PathIndexResolverUnitTests.java
index 3d51e3445..c9f96b2ad 100644
--- a/src/test/java/org/springframework/data/redis/core/convert/PathIndexResolverUnitTests.java
+++ b/src/test/java/org/springframework/data/redis/core/convert/PathIndexResolverUnitTests.java
@@ -32,6 +32,7 @@ import java.util.Map;
 import java.util.Set;
 
 import org.hamcrest.core.IsCollectionContaining;
+import org.hamcrest.core.IsInstanceOf;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -223,6 +224,7 @@ public class PathIndexResolverUnitTests {
 
 	/**
 	 * @see DATAREDIS-425
+	 * @see DATAREDIS-471
 	 */
 	@Test
 	public void shouldIgnoreConfiguredIndexesInMapWhenValueIsNull() {
@@ -235,7 +237,8 @@ public class PathIndexResolverUnitTests {
 
 		Set<IndexedData> indexes = indexResolver.resolveIndexesFor(ClassTypeInformation.from(Person.class), rand);
 
-		assertThat(indexes.size(), is(0));
+		assertThat(indexes.size(), is(1));
+		assertThat(indexes.iterator().next(), IsInstanceOf.instanceOf(RemoveIndexedData.class));
 	}
 
 	/**
diff --git a/src/test/java/org/springframework/data/redis/core/mapping/ConfigAwareTimeToLiveAccessorUnitTests.java b/src/test/java/org/springframework/data/redis/core/mapping/ConfigAwareTimeToLiveAccessorUnitTests.java
index 8e85f85d4..656183b04 100644
--- a/src/test/java/org/springframework/data/redis/core/mapping/ConfigAwareTimeToLiveAccessorUnitTests.java
+++ b/src/test/java/org/springframework/data/redis/core/mapping/ConfigAwareTimeToLiveAccessorUnitTests.java
@@ -21,6 +21,7 @@ import static org.junit.Assert.*;
 
 import org.junit.Before;
 import org.junit.Test;
+import org.springframework.data.redis.core.PartialUpdate;
 import org.springframework.data.redis.core.RedisHash;
 import org.springframework.data.redis.core.TimeToLive;
 import org.springframework.data.redis.core.convert.KeyspaceConfiguration;
@@ -176,6 +177,55 @@ public class ConfigAwareTimeToLiveAccessorUnitTests {
 		assertThat(accessor.getTimeToLive(new TypeWithTtlOnMethod(100L)), is(100L));
 	}
 
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void getTimeToLiveShouldReturnDefaultValue() {
+
+		Long ttl = accessor
+				.getTimeToLive(new PartialUpdate<TypeWithRedisHashAnnotation>("123", new TypeWithRedisHashAnnotation()));
+
+		assertThat(ttl, is(5L));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void getTimeToLiveShouldReturnValueWhenUpdateModifiesTtlProperty() {
+
+		Long ttl = accessor
+				.getTimeToLive(new PartialUpdate<SimpleTypeWithTTLProperty>("123", new SimpleTypeWithTTLProperty())
+						.set("ttl", 100).refreshTtl(true));
+
+		assertThat(ttl, is(100L));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void getTimeToLiveShouldReturnPropertyValueWhenUpdateModifiesTtlProperty() {
+
+		Long ttl = accessor.getTimeToLive(new PartialUpdate<TypeWithRedisHashAnnotationAndTTLProperty>("123",
+				new TypeWithRedisHashAnnotationAndTTLProperty()).set("ttl", 100).refreshTtl(true));
+
+		assertThat(ttl, is(100L));
+	}
+
+	/**
+	 * @see DATAREDIS-471
+	 */
+	@Test
+	public void getTimeToLiveShouldReturnDefaultValueWhenUpdateDoesNotModifyTtlProperty() {
+
+		Long ttl = accessor.getTimeToLive(new PartialUpdate<TypeWithRedisHashAnnotationAndTTLProperty>("123",
+				new TypeWithRedisHashAnnotationAndTTLProperty()).refreshTtl(true));
+
+		assertThat(ttl, is(10L));
+	}
+
 	static class SimpleType {}
 
 	static class SimpleTypeWithTTLProperty {