diff --git a/pom.xml b/pom.xml
index e6d868411f..556ad21853 100644
--- a/pom.xml
+++ b/pom.xml
@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-redis</artifactId>
-	<version>1.8.0.BUILD-SNAPSHOT</version>
+	<version>1.8.0.DATAREDIS-471-SNAPSHOT</version>
 
 	<name>Spring Data Redis</name>
 
diff --git a/src/main/asciidoc/reference/redis-repositories.adoc b/src/main/asciidoc/reference/redis-repositories.adoc
index 9cca5cbdb9..9039d4f24a 100644
--- a/src/main/asciidoc/reference/redis-repositories.adoc
+++ b/src/main/asciidoc/reference/redis-repositories.adoc
@@ -487,6 +487,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` and `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 e25a7a52e8..054ca9c27d 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.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 @@
  * 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 @@ private void createOrUpdateIndexes(Object key, Iterable<IndexedData> indexValues
 
 	/**
 	 * Removes a key from all available indexes.
-	 * 
+	 *
 	 * @param key must not be {@literal null}.
 	 */
 	public void removeKeyFromIndexes(String keyspace, Object key) {
@@ -142,13 +143,14 @@ private void removeKeyFromExistingIndexes(byte[] key, Iterable<IndexedData> inde
 
 	/**
 	 * Remove given key from all indexes matching {@link IndexedData#getIndexName()}:
-	 * 
+	 *
 	 * @param key
 	 * @param indexedData
 	 */
 	protected void removeKeyFromExistingIndexes(byte[] key, IndexedData indexedData) {
 
 		Assert.notNull(indexedData, "IndexedData must not be null!");
+
 		Set<byte[]> existingKeys = connection
 				.keys(toBytes(indexedData.getKeyspace() + ":" + indexedData.getIndexName() + ":*"));
 
@@ -168,7 +170,7 @@ private void addKeyToIndexes(byte[] key, Iterable<IndexedData> indexValues) {
 
 	/**
 	 * 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 +179,11 @@ protected void addKeyToIndex(byte[] key, IndexedData indexedData) {
 		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();
 
@@ -212,7 +218,8 @@ private byte[] toBytes(Object source) {
 		}
 
 		throw new InvalidDataAccessApiUsageException(String.format(
-				"Cannot convert %s to binary representation for index key generation. Are you missing a Converter? Did you register a non PathBasedRedisIndexDefinition that might apply to a complex type?",
+				"Cannot convert %s to binary representation for index key generation. "
+						+ "Are you missing a Converter? Did you register a non PathBasedRedisIndexDefinition that might apply to a complex type?",
 				source.getClass()));
 	}
 
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 0000000000..a8913614f2
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/core/PartialUpdate.java
@@ -0,0 +1,238 @@
+/*
+ * 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 481d7463f3..65c59e433f 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 @@
 
 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;
@@ -37,6 +39,8 @@
 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;
@@ -45,6 +49,7 @@
 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;
@@ -56,7 +61,7 @@
  * are stored in a Redis Hash using the value of {@link RedisHash}, the {@link KeyspaceConfiguration} or just
  * {@link Class#getName()} as a prefix. <br />
  * <strong>Example</strong>
- * 
+ *
  * <pre>
  * <code>
  * &#64;RedisHash("persons")
@@ -64,8 +69,8 @@
  *   &#64;Id String id;
  *   String name;
  * }
- * 
- * 
+ *
+ *
  *         prefix              ID
  *           |                 |
  *           V                 V
@@ -76,11 +81,11 @@
  * 4) Rand al'Thor
  * </code>
  * </pre>
- * 
+ *
  * <br />
  * The {@link KeyValueAdapter} is <strong>not</strong> intended to store simple types such as {@link String} values.
  * Please use {@link RedisTemplate} for this purpose.
- * 
+ *
  * @author Christoph Strobl
  * @author Mark Paluch
  * @since 1.7
@@ -98,7 +103,7 @@ public class RedisKeyValueAdapter extends AbstractKeyValueAdapter
 	/**
 	 * Creates new {@link RedisKeyValueAdapter} with default {@link RedisMappingContext} and default
 	 * {@link CustomConversions}.
-	 * 
+	 *
 	 * @param redisOps must not be {@literal null}.
 	 */
 	public RedisKeyValueAdapter(RedisOperations<?, ?> redisOps) {
@@ -107,7 +112,7 @@ public RedisKeyValueAdapter(RedisOperations<?, ?> redisOps) {
 
 	/**
 	 * Creates new {@link RedisKeyValueAdapter} with default {@link CustomConversions}.
-	 * 
+	 *
 	 * @param redisOps must not be {@literal null}.
 	 * @param mappingContext must not be {@literal null}.
 	 */
@@ -117,7 +122,7 @@ public RedisKeyValueAdapter(RedisOperations<?, ?> redisOps, RedisMappingContext
 
 	/**
 	 * Creates new {@link RedisKeyValueAdapter}.
-	 * 
+	 *
 	 * @param redisOps must not be {@literal null}.
 	 * @param mappingContext must not be {@literal null}.
 	 * @param customConversions can be {@literal null}.
@@ -143,7 +148,7 @@ public RedisKeyValueAdapter(RedisOperations<?, ?> redisOps, RedisMappingContext
 
 	/**
 	 * Creates new {@link RedisKeyValueAdapter} with specific {@link RedisConverter}.
-	 * 
+	 *
 	 * @param redisOps must not be {@literal null}.
 	 * @param mappingContext must not be {@literal null}.
 	 */
@@ -395,9 +400,124 @@ public Long doInRedis(RedisConnection connection) throws DataAccessException {
 		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}.
-	 * 
+	 *
 	 * @param callback must not be {@literal null}.
 	 * @see RedisOperations#execute(RedisCallback)
 	 * @return
@@ -408,7 +528,7 @@ public <T> T execute(RedisCallback<T> callback) {
 
 	/**
 	 * Get the {@link RedisConverter} in use.
-	 * 
+	 *
 	 * @return never {@literal null}.
 	 */
 	public RedisConverter getConverter() {
@@ -430,7 +550,7 @@ public byte[] createKey(String keyspace, String id) {
 
 	/**
 	 * Convert given source to binary representation using the underlying {@link ConversionService}.
-	 * 
+	 *
 	 * @param source
 	 * @return
 	 * @throws ConverterNotFoundException
@@ -509,7 +629,7 @@ private void initKeyExpirationListener() {
 	 * {@link MessageListener} implementation used to capture Redis keypspace notifications. Tries to read a previously
 	 * created phantom key {@code keyspace:id:phantom} to provide the expired object as part of the published
 	 * {@link RedisKeyExpiredEvent}.
-	 * 
+	 *
 	 * @author Christoph Strobl
 	 * @since 1.7
 	 */
@@ -520,7 +640,7 @@ static class MappingExpirationListener extends KeyExpirationEventMessageListener
 
 		/**
 		 * Creates new {@link MappingExpirationListener}.
-		 * 
+		 *
 		 * @param listenerContainer
 		 * @param ops
 		 * @param converter
@@ -582,4 +702,33 @@ private boolean isKeyExpirationMessage(Message message) {
 		}
 	}
 
+	/**
+	 * 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 2f1d1f057c..8a1f86e615 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 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 RedisMappingContext getMappingContext() {
 	 * </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 List<T> doInRedis(RedisKeyValueAdapter adapter) {
 					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,47 @@ public List<T> doInRedis(RedisKeyValueAdapter adapter) {
 		});
 	}
 
+	/*
+	 * (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);
+	}
+
+	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 2ebe21b3f7..7ef63c103d 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,18 @@ public Set<IndexedData> resolveIndexesFor(TypeInformation<?> typeInformation, Ob
 		return data;
 	}
 
+	/* (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) {
+
+		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 a90e376f7a..ce79e6c409 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
@@ -23,7 +23,7 @@
 /**
  * {@link IndexResolver} extracts secondary index structures to be applied on a given path, {@link PersistentProperty}
  * and value.
- * 
+ *
  * @author Christoph Strobl
  * @since 1.7
  */
@@ -31,11 +31,22 @@ public interface IndexResolver {
 
 	/**
 	 * Resolves all indexes for given type information / value combination.
-	 * 
+	 *
 	 * @param typeInformation must not be {@literal null}.
 	 * @param value the actual value. Can be {@literal null}.
 	 * @return never {@literal null}.
 	 */
 	Set<IndexedData> resolveIndexesFor(TypeInformation<?> typeInformation, Object value);
 
+	/**
+	 * Resolves all indexes for given type information / value combination.
+	 *
+	 * @param keyspace must not be {@literal null}.
+	 * @param path must not be {@literal null}.
+	 * @param typeInformation must not be {@literal null}.
+	 * @param value the actual value. Can be {@literal null}.
+	 * @return never {@literal null}.
+	 */
+	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 c503e71fea..4fa693bf45 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.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;
@@ -68,18 +72,18 @@
  * <br />
  * <strong>NOTE</strong> {@link MappingRedisConverter} is an {@link InitializingBean} and requires
  * {@link MappingRedisConverter#afterPropertiesSet()} to be called.
- * 
+ *
  * <pre>
  * <code>
  * &#64;RedisHash("persons")
  * class Person {
- * 
+ *
  *   &#64;Id String id;
  *   String firstname;
- * 
+ *
  *   List<String> nicknames;
  *   List<Person> coworkers;
- * 
+ *
  *   Address address;
  *   &#64;Reference Country nationality;
  * }
@@ -101,14 +105,16 @@
  * nationality=nationality:andora
  * </code>
  * </pre>
- * 
+ *
  * @author Christoph Strobl
  * @author Greg Turnquist
+ * @author Mark Paluch
  * @since 1.7
  */
 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;
@@ -123,7 +129,7 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 
 	/**
 	 * Creates new {@link MappingRedisConverter}.
-	 * 
+	 *
 	 * @param context can be {@literal null}.
 	 */
 	MappingRedisConverter(RedisMappingContext context) {
@@ -132,7 +138,7 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean {
 
 	/**
 	 * Creates new {@link MappingRedisConverter} and defaults {@link RedisMappingContext} when {@literal null}.
-	 * 
+	 *
 	 * @param mappingContext can be {@literal null}.
 	 * @param indexResolver can be {@literal null}.
 	 * @param referenceResolver must not be {@literal null}.
@@ -223,12 +229,18 @@ public void doWithPersistentProperty(KeyValuePersistentProperty persistentProper
 
 				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 +248,9 @@ else if (persistentProperty.isCollectionLike()) {
 
 					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 +353,15 @@ public void doWithAssociation(Association<KeyValuePersistentProperty> associatio
 	@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 +388,134 @@ public void write(Object source, final RedisData sink) {
 		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(TYPE_HINT_ALIAS)) {
+			sink.getBucket().put(TYPE_HINT_ALIAS, 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())) {
+				writePartialPropertyUpdate(update, pUpdate, sink, entity, path);
+			}
+		}
+	}
+
+	/**
+	 * @param update
+	 * @param pUpdate
+	 * @param sink
+	 * @param entity
+	 * @param path
+	 */
+	private void writePartialPropertyUpdate(PartialUpdate<?> update, PropertyUpdate pUpdate, RedisData sink,
+			RedisPersistentEntity<?> entity, String path) {
+
+		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);
+			return;
+		}
+
+		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 Entry) {
+				map.put(((Entry<?, ?>) pUpdate.getValue()).getKey(), ((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 +537,11 @@ private void writeInternal(final String keyspace, final String path, final Objec
 			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;
@@ -444,10 +599,10 @@ public void doWithPersistentProperty(KeyValuePersistentProperty persistentProper
 			}
 		});
 
-		writeAssiciation(keyspace, path, entity, value, sink);
+		writeAssociation(path, entity, value, sink);
 	}
 
-	private void writeAssiciation(final String keyspace, final String path, final KeyValuePersistentEntity<?> entity,
+	private void writeAssociation(final String path, final KeyValuePersistentEntity<?> entity,
 			final Object value, final RedisData sink) {
 
 		if (value == null) {
@@ -520,6 +675,11 @@ private void writeCollection(String keyspace, String path, Iterable<?> values, T
 
 			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 +752,7 @@ private Object readCollectionOrArray(String path, Class<?> collectionType, Class
 			}
 		}
 
-		return isArray ? toArray(target, collectionType, valueType) : target;
+		return isArray ? toArray(target, collectionType, valueType) : (target.isEmpty() ? null : target);
 	}
 
 	/**
@@ -616,6 +776,11 @@ private void writeMap(String keyspace, String path, Class<?> mapValueType, Map<?
 
 			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 +824,7 @@ private void writeMap(String keyspace, String path, Class<?> mapValueType, Map<?
 			target.put(key, fromBytes(entry.getValue(), typeToUse));
 		}
 
-		return target;
+		return target.isEmpty() ? null : target;
 	}
 
 	/**
@@ -700,7 +865,7 @@ private void writeMap(String keyspace, String path, Class<?> mapValueType, Map<?
 			target.put(mapKey, o);
 		}
 
-		return target;
+		return target.isEmpty() ? null : target;
 	}
 
 	private Class<?> getTypeHint(String path, Bucket bucket, Class<?> fallback) {
@@ -723,7 +888,7 @@ private Class<?> getTypeHint(String path, Bucket bucket, Class<?> fallback) {
 
 	/**
 	 * Convert given source to binary representation using the underlying {@link ConversionService}.
-	 * 
+	 *
 	 * @param source
 	 * @return
 	 * @throws ConverterNotFoundException
@@ -739,7 +904,7 @@ public byte[] toBytes(Object source) {
 
 	/**
 	 * Convert given binary representation to desired target type using the underlying {@link ConversionService}.
-	 * 
+	 *
 	 * @param source
 	 * @param type
 	 * @return
@@ -759,6 +924,10 @@ public <T> T fromBytes(byte[] source, Class<T> type) {
 	 */
 	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,12 +939,12 @@ private Object toArray(Collection<Object> source, Class<?> arrayType, Class<?> v
 			Array.set(targetArray, i, conversionService.convert(iterator.next(), valueType));
 			i++;
 		}
-		return targetArray;
+		return i > 0 ? targetArray : null;
 	}
 
 	/**
 	 * Set {@link CustomConversions} to be applied.
-	 * 
+	 *
 	 * @param customConversions
 	 */
 	public void setCustomConversions(CustomConversions customConversions) {
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 eeaf6a931a..6498663948 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.util.ClassTypeInformation;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
 import org.springframework.util.CollectionUtils;
 
 /**
@@ -62,7 +62,7 @@ public PathIndexResolver() {
 	/**
 	 * Creates new {@link PathIndexResolver} with given {@link IndexConfiguration}.
 	 *
-	 * @param mapppingContext must not be {@literal null}.
+	 * @param mappingContext must not be {@literal null}.
 	 */
 	public PathIndexResolver(RedisMappingContext mappingContext) {
 
@@ -80,6 +80,15 @@ public Set<IndexedData> resolveIndexesFor(TypeInformation<?> typeInformation, Ob
 				null, value);
 	}
 
+	/* (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 doResolveIndexesFor(keyspace, path, typeInformation, null, value);
+	}
+
 	private Set<IndexedData> doResolveIndexesFor(final String keyspace, final String path,
 			TypeInformation<?> typeInformation, PersistentProperty<?> fallback, Object value) {
 
@@ -89,6 +98,13 @@ private Set<IndexedData> doResolveIndexesFor(final String keyspace, final String
 			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>();
 
@@ -121,7 +137,7 @@ public void doWithPersistentProperty(KeyValuePersistentProperty persistentProper
 						final Iterable<?> iterable;
 
 						if (Iterable.class.isAssignableFrom(propertyValue.getClass())) {
-							iterable = (Iterable) propertyValue;
+							iterable = (Iterable<?>) propertyValue;
 						} else if (propertyValue.getClass().isArray()) {
 							iterable = CollectionUtils.arrayToList(propertyValue);
 						} else {
@@ -173,10 +189,6 @@ private TypeInformation<?> updateTypeHintForActualValue(TypeInformation<?> typeH
 	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 +204,14 @@ protected Set<IndexedData> resolveIndex(String keyspace, String propertyPath, Pe
 					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 +222,7 @@ else if (property != null && property.isAnnotationPresent(Indexed.class)) {
 
 			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 5754da8a87..bf3c945a1a 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
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-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.
@@ -18,14 +18,21 @@
 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}.
- * 
+ *
  * @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 0bfd78b6c8..e3a7004877 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
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-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.
@@ -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;
@@ -26,7 +27,7 @@
 /**
  * Data object holding {@link Bucket} representing the domain object to be stored in a Redis hash. Index information
  * points to additional structures holding the objects is for searching.
- * 
+ *
  * @author Christoph Strobl
  * @since 1.7
  */
@@ -49,7 +50,7 @@ public RedisData() {
 
 	/**
 	 * Creates new {@link RedisData} with {@link Bucket} holding provided values.
-	 * 
+	 *
 	 * @param raw should not be {@literal null}.
 	 */
 	public RedisData(Map<byte[], byte[]> raw) {
@@ -58,7 +59,7 @@ public RedisData(Map<byte[], byte[]> raw) {
 
 	/**
 	 * Creates new {@link RedisData} with {@link Bucket}
-	 * 
+	 *
 	 * @param bucket must not be {@literal null}.
 	 */
 	public RedisData(Bucket bucket) {
@@ -70,7 +71,7 @@ public RedisData(Bucket bucket) {
 
 	/**
 	 * Set the id to be used as part of the key.
-	 * 
+	 *
 	 * @param id
 	 */
 	public void setId(String id) {
@@ -86,7 +87,7 @@ public String getId() {
 
 	/**
 	 * Get the time before expiration in seconds.
-	 * 
+	 *
 	 * @return {@literal null} if not set.
 	 */
 	public Long getTimeToLive() {
@@ -94,7 +95,7 @@ public Long getTimeToLive() {
 	}
 
 	/**
-	 * @param index
+	 * @param index must not be {@literal null}.
 	 */
 	public void addIndexedData(IndexedData index) {
 
@@ -102,6 +103,15 @@ public void addIndexedData(IndexedData index) {
 		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}.
 	 */
@@ -132,7 +142,7 @@ public Bucket getBucket() {
 
 	/**
 	 * Set the time before expiration in {@link TimeUnit#SECONDS}.
-	 * 
+	 *
 	 * @param timeToLive can be {@literal null}.
 	 */
 	public void setTimeToLive(Long timeToLive) {
@@ -141,7 +151,7 @@ public void setTimeToLive(Long timeToLive) {
 
 	/**
 	 * Set the time before expiration converting the given arguments to {@link TimeUnit#SECONDS}.
-	 * 
+	 *
 	 * @param timeToLive must not be {@literal null}
 	 * @param timeUnit must not be {@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 0000000000..b50cf149ca
--- /dev/null
+++ b/src/main/java/org/springframework/data/redis/core/convert/RemoveIndexedData.java
@@ -0,0 +1,56 @@
+/*
+ * 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;
+
+/**
+ * {@link RemoveIndexedData} represents a removed index entry from a secondary index for a property path in a given keyspace.
+ *
+ * @author Christoph Strobl
+ * @author Mark Paluch
+ */
+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 b4203f243e..d8cd80fe65 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,15 @@ public Set<IndexedData> resolveIndexesFor(TypeInformation<?> typeInformation, Ob
 		return 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 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 3017bda5fe..f4c333fd26 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.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.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 RedisMappingContext() {
 	 */
 	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 @@ static class ConfigAwareTimeToLiveAccessor implements TimeToLiveAccessor {
 		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 Long getTimeToLive(final Object source) {
 				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 Long getTimeToLive(final Object source) {
 							return TimeUnit.SECONDS.convert(timeout.longValue(), ttl.unit());
 						}
 					} catch (IllegalAccessException e) {
-						throw new IllegalStateException("Not allowed to access method '" + timeoutMethod.getName() + "': "
-								+ e.getMessage(), 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);
+								"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 fb235512aa..96d38ace8d 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 @@ private static class NoOpReferenceResolver implements ReferenceResolver {
 
 		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 @@ private static class NoOpIndexResolver implements IndexResolver {
 
 		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 8084ee8d5c..fdbec9e210 100644
--- a/src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterTests.java
+++ b/src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterTests.java
@@ -15,28 +15,33 @@
  */
 package org.springframework.data.redis.core;
 
-import static org.hamcrest.core.Is.*;
-import static org.hamcrest.core.IsCollectionContaining.*;
-import static org.hamcrest.core.IsInstanceOf.*;
-import static org.hamcrest.core.IsNot.*;
+import static org.hamcrest.Matchers.*;
 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 +53,35 @@
  * @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();
 
@@ -88,14 +109,6 @@ public void tearDown() {
 		} catch (Exception e) {
 			// ignore
 		}
-
-		try {
-			if (connectionFactory instanceof DisposableBean) {
-				((DisposableBean) connectionFactory).destroy();
-			}
-		} catch (Exception e) {
-			// ignore
-		}
 	}
 
 	/**
@@ -311,6 +324,191 @@ public void putWritesIndexDataCorrectly() {
 		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/RedisKeyValueAdapterUnitTests.java b/src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterUnitTests.java
index 6783ed65e4..90dd67adda 100644
--- a/src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterUnitTests.java
+++ b/src/test/java/org/springframework/data/redis/core/RedisKeyValueAdapterUnitTests.java
@@ -16,7 +16,6 @@
 
 package org.springframework.data.redis.core;
 
-import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
 
 import java.util.Arrays;
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 88d69c5670..3507b22581 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 @@
 
 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.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 static List<RedisConnectionFactory> params() {
 		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 void setUp() {
 		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,9 @@ public Void doInRedis(RedisConnection connection) throws DataAccessException {
 				return null;
 			}
 		});
+
+		template.destroy();
+		adapter.destroy();
 	}
 
 	/**
@@ -198,16 +214,650 @@ public List<byte[]> doInRedis(RedisConnection connection) throws DataAccessExcep
 		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 +878,25 @@ public boolean equals(Object obj) {
 				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 867c58a141..683660f4d8 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
@@ -16,10 +16,7 @@
 package org.springframework.data.redis.core.convert;
 
 import static org.hamcrest.collection.IsIterableContainingInOrder.contains;
-import static org.hamcrest.core.Is.*;
-import static org.hamcrest.core.IsCollectionContaining.*;
-import static org.hamcrest.core.IsInstanceOf.*;
-import static org.hamcrest.core.IsNull.*;
+import static org.hamcrest.Matchers.*;
 import static org.junit.Assert.*;
 import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
@@ -49,13 +46,17 @@
 
 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 +85,7 @@
 @RunWith(MockitoJUnitRunner.class)
 public class MappingRedisConverterUnitTests {
 
+	public @Rule ExpectedException exception = ExpectedException.none();
 	@Mock ReferenceResolver resolverMock;
 	MappingRedisConverter converter;
 	Person rand;
@@ -95,7 +97,6 @@ public void setUp() {
 		converter.afterPropertiesSet();
 
 		rand = new Person();
-
 	}
 
 	/**
@@ -1582,6 +1583,7 @@ public void shouldWriteReadObjectListValueTypeCorrectly() {
 	}
 
 	/**
+	 *
 	 * @see DATAREDIS-509
 	 */
 	@Test
@@ -1611,6 +1613,405 @@ public void readHandlesArraysOfPrimitivesProperly() {
 				.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 3d51e3445c..c9f96b2ada 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.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 void shouldResolveConfiguredIndexesInMapOfComplexTypes() {
 
 	/**
 	 * @see DATAREDIS-425
+	 * @see DATAREDIS-471
 	 */
 	@Test
 	public void shouldIgnoreConfiguredIndexesInMapWhenValueIsNull() {
@@ -235,7 +237,8 @@ public void shouldIgnoreConfiguredIndexesInMapWhenValueIsNull() {
 
 		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 8e85f85d41..321c193b24 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
@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-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.
@@ -15,12 +15,12 @@
  */
 package org.springframework.data.redis.core.mapping;
 
-import static org.hamcrest.core.Is.*;
-import static org.hamcrest.core.IsNull.*;
+import static org.hamcrest.Matchers.*;
 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 +176,55 @@ public void getTimeToLiveShouldReturnValueWhenMethodLevelTimeToLiveIfPresentAlth
 		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 {