DATACASS-523 - Extend converter to map TupleValue to domain objects.

We now support mapped tuple values via @Tuple and @Element(1) annotations. Mapped tuple values can be embedded within table entities and user-defined. Mapped tuples can contain user-defined types, collection types and simple property types and are supported with schema generation.

	@Tuple
	class Address {

		@Element(0) String city;

		@Element(1) String street;

		@Element(2) int sortOrder;
	}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/AbstractCassandraConfiguration.java

@@ -30,6 +30,7 @@ import org.springframework.data.cassandra.core.convert.CassandraCustomConversion
 import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
 import org.springframework.data.cassandra.core.cql.session.DefaultSessionFactory;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
+import org.springframework.data.cassandra.core.mapping.SimpleTupleTypeFactory;
 import org.springframework.data.cassandra.core.mapping.SimpleUserTypeResolver;
 import org.springframework.data.cassandra.core.mapping.Table;
 import org.springframework.data.convert.CustomConversions;
@@ -37,6 +38,7 @@ import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
+import com.datastax.driver.core.Cluster;
 import com.datastax.driver.core.Session;
 
 /**
@@ -139,7 +141,10 @@ public abstract class AbstractCassandraConfiguration extends AbstractClusterConf
 	@Bean
 	public CassandraMappingContext cassandraMapping() throws ClassNotFoundException {
 
-		CassandraMappingContext mappingContext = new CassandraMappingContext();
+		Cluster cluster = getRequiredCluster();
+
+		CassandraMappingContext mappingContext = new CassandraMappingContext(
+				new SimpleUserTypeResolver(cluster, getKeyspaceName()), new SimpleTupleTypeFactory(cluster));
 
 		if (beanClassLoader != null) {
 			mappingContext.setBeanClassLoader(beanClassLoader);
@@ -151,7 +156,6 @@ public abstract class AbstractCassandraConfiguration extends AbstractClusterConf
 
 		mappingContext.setCustomConversions(customConversions);
 		mappingContext.setSimpleTypeHolder(customConversions.getSimpleTypeHolder());
-		mappingContext.setUserTypeResolver(new SimpleUserTypeResolver(getRequiredCluster(), getKeyspaceName()));
 
 		return mappingContext;
 	}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/BasicCassandraRowValueProvider.java

@@ -37,20 +37,33 @@ public class BasicCassandraRowValueProvider implements CassandraRowValueProvider
 
 	private final SpELExpressionEvaluator evaluator;
 
+	/**
+	 * Create a new {@link BasicCassandraRowValueProvider} with the given {@link Row} and {@link SpELExpressionEvaluator}.
+	 *
+	 * @param source must not be {@literal null}.
+	 * @param evaluator must not be {@literal null}.
+	 * @since 2.1
+	 */
+	public BasicCassandraRowValueProvider(Row source, SpELExpressionEvaluator evaluator) {
+
+		Assert.notNull(source, "Source Row must not be null");
+		Assert.notNull(evaluator, "SpELExpressionEvaluator must not be null");
+
+		this.reader = new ColumnReader(source);
+		this.evaluator = evaluator;
+	}
+
 	/**
 	 * Create a new {@link BasicCassandraRowValueProvider} with the given {@link Row} and
 	 * {@link DefaultSpELExpressionEvaluator}.
 	 *
 	 * @param source must not be {@literal null}.
 	 * @param evaluator must not be {@literal null}.
+	 * @deprecated since 2.1, use {@link #BasicCassandraRowValueProvider(Row, SpELExpressionEvaluator)}
 	 */
+	@Deprecated
 	public BasicCassandraRowValueProvider(Row source, DefaultSpELExpressionEvaluator evaluator) {
-
-		Assert.notNull(source, "Source Row must not be null");
-		Assert.notNull(evaluator, "DefaultSpELExpressionEvaluator must not be null");
-
-		this.reader = new ColumnReader(source);
-		this.evaluator = evaluator;
+		this(source, (SpELExpressionEvaluator) evaluator);
 	}
 
 	/* (non-Javadoc)
@@ -66,7 +79,7 @@ public class BasicCassandraRowValueProvider implements CassandraRowValueProvider
 			return evaluator.evaluate(spelExpression);
 		}
 
-		return (T) reader.get(property.getColumnName());
+		return (T) reader.get(property.getRequiredColumnName());
 	}
 
 	/* (non-Javadoc)
@@ -85,6 +98,6 @@ public class BasicCassandraRowValueProvider implements CassandraRowValueProvider
 
 		Assert.notNull(property, "CassandraPersistentProperty must not be null");
 
-		return getRow().getColumnDefinitions().contains(property.getColumnName().toCql());
+		return getRow().getColumnDefinitions().contains(property.getRequiredColumnName().toCql());
 	}
 }

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/CassandraTupleValueProvider.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2018 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.cassandra.core.convert;
+
+import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
+import org.springframework.data.mapping.model.SpELExpressionEvaluator;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+import com.datastax.driver.core.CodecRegistry;
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.TupleValue;
+
+/**
+ * {@link CassandraValueProvider} to read property values from a {@link TupleValue}.
+ *
+ * @author Mark Paluch
+ * @since 2.1
+ */
+public class CassandraTupleValueProvider implements CassandraValueProvider {
+
+	private final TupleValue tupleValue;
+
+	private final CodecRegistry codecRegistry;
+
+	private final SpELExpressionEvaluator evaluator;
+
+	/**
+	 * Create a new {@link CassandraTupleValueProvider} with the given {@link TupleValue} and
+	 * {@link SpELExpressionEvaluator}.
+	 *
+	 * @param tupleValue must not be {@literal null}.
+	 * @param codecRegistry must not be {@literal null}.
+	 * @param evaluator must not be {@literal null}.
+	 */
+	public CassandraTupleValueProvider(TupleValue tupleValue, CodecRegistry codecRegistry,
+			SpELExpressionEvaluator evaluator) {
+
+		Assert.notNull(tupleValue, "TupleValue must not be null");
+		Assert.notNull(codecRegistry, "CodecRegistry must not be null");
+		Assert.notNull(evaluator, "SpELExpressionEvaluator must not be null");
+
+		this.tupleValue = tupleValue;
+		this.codecRegistry = codecRegistry;
+		this.evaluator = evaluator;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.convert.CassandraValueProvider#hasProperty(org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty)
+	 */
+	@Override
+	public boolean hasProperty(CassandraPersistentProperty property) {
+		return tupleValue.getType().getComponentTypes().size() >= property.getRequiredOrdinal();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mapping.model.PropertyValueProvider#getPropertyValue(org.springframework.data.mapping.PersistentProperty)
+	 */
+	@Nullable
+	@Override
+	public <T> T getPropertyValue(CassandraPersistentProperty property) {
+
+		String spelExpression = property.getSpelExpression();
+		if (spelExpression != null) {
+			return evaluator.evaluate(spelExpression);
+		}
+
+		int ordinal = property.getRequiredOrdinal();
+		DataType elementType = tupleValue.getType().getComponentTypes().get(ordinal);
+
+		return tupleValue.get(ordinal, codecRegistry.codecFor(elementType));
+	}
+}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/CassandraUDTValueProvider.java

@@ -40,15 +40,15 @@ public class CassandraUDTValueProvider implements CassandraValueProvider {
 	private final SpELExpressionEvaluator evaluator;
 
 	/**
-	 * Create a new {@link CassandraUDTValueProvider} with the given {@link UDTValue} and
-	 * {@link DefaultSpELExpressionEvaluator}.
+	 * Create a new {@link CassandraUDTValueProvider} with the given {@link UDTValue} and {@link SpELExpressionEvaluator}.
 	 *
 	 * @param udtValue must not be {@literal null}.
 	 * @param codecRegistry must not be {@literal null}.
 	 * @param evaluator must not be {@literal null}.
+	 * @since 2.1
 	 */
 	public CassandraUDTValueProvider(UDTValue udtValue, CodecRegistry codecRegistry,
-			DefaultSpELExpressionEvaluator evaluator) {
+			SpELExpressionEvaluator evaluator) {
 
 		Assert.notNull(udtValue, "UDTValue must not be null");
 		Assert.notNull(codecRegistry, "CodecRegistry must not be null");
@@ -59,6 +59,21 @@ public class CassandraUDTValueProvider implements CassandraValueProvider {
 		this.evaluator = evaluator;
 	}
 
+	/**
+	 * Create a new {@link CassandraUDTValueProvider} with the given {@link UDTValue} and
+	 * {@link DefaultSpELExpressionEvaluator}.
+	 *
+	 * @param udtValue must not be {@literal null}.
+	 * @param codecRegistry must not be {@literal null}.
+	 * @param evaluator must not be {@literal null}.
+	 * @deprecated since 2.1, use {@link #CassandraUDTValueProvider(UDTValue, CodecRegistry, SpELExpressionEvaluator)}
+	 */
+	@Deprecated
+	public CassandraUDTValueProvider(UDTValue udtValue, CodecRegistry codecRegistry,
+			DefaultSpELExpressionEvaluator evaluator) {
+		this(udtValue, codecRegistry, (SpELExpressionEvaluator) evaluator);
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.mapping.model.PropertyValueProvider#getPropertyValue(org.springframework.data.mapping.PersistentProperty)
 	 */
@@ -71,7 +86,7 @@ public class CassandraUDTValueProvider implements CassandraValueProvider {
 			return evaluator.evaluate(spelExpression);
 		}
 
-		String name = property.getColumnName().toCql();
+		String name = property.getRequiredColumnName().toCql();
 		DataType fieldType = udtValue.getType().getFieldType(name);
 
 		return udtValue.get(name, codecRegistry.codecFor(fieldType));
@@ -82,6 +97,6 @@ public class CassandraUDTValueProvider implements CassandraValueProvider {
 	 */
 	@Override
 	public boolean hasProperty(CassandraPersistentProperty property) {
-		return udtValue.getType().contains(property.getColumnName().toCql());
+		return udtValue.getType().contains(property.getRequiredColumnName().toCql());
 	}
 }

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverter.java

@@ -15,15 +15,18 @@
  */
 package org.springframework.data.cassandra.core.convert;
 
+import lombok.AllArgsConstructor;
+
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.function.Function;
 
-import lombok.AllArgsConstructor;
-
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.context.ApplicationContext;
@@ -47,6 +50,7 @@ import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
 import org.springframework.data.mapping.model.DefaultSpELExpressionEvaluator;
 import org.springframework.data.mapping.model.PersistentEntityParameterValueProvider;
 import org.springframework.data.mapping.model.SpELContext;
+import org.springframework.data.mapping.model.SpELExpressionEvaluator;
 import org.springframework.data.util.ClassTypeInformation;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.lang.Nullable;
@@ -54,12 +58,11 @@ import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.ObjectUtils;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import com.datastax.driver.core.CodecRegistry;
 import com.datastax.driver.core.DataType;
 import com.datastax.driver.core.Row;
+import com.datastax.driver.core.TupleType;
+import com.datastax.driver.core.TupleValue;
 import com.datastax.driver.core.TypeCodec;
 import com.datastax.driver.core.UDTValue;
 import com.datastax.driver.core.UserType;
@@ -175,39 +178,35 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			return getConversionService().convert(row, type);
 		}
 
-		CassandraPersistentEntity<R> persistentEntity =
-				(CassandraPersistentEntity<R>) getMappingContext().getRequiredPersistentEntity(typeInfo);
+		CassandraPersistentEntity<R> persistentEntity = (CassandraPersistentEntity<R>) getMappingContext()
+				.getRequiredPersistentEntity(typeInfo);
 
 		return readEntityFromRow(persistentEntity, row);
 	}
 
 	protected <S> S readEntityFromRow(CassandraPersistentEntity<S> entity, Row row) {
-
-		DefaultSpELExpressionEvaluator expressionEvaluator = new DefaultSpELExpressionEvaluator(row, spELContext);
-
-		BasicCassandraRowValueProvider rowValueProvider = new BasicCassandraRowValueProvider(row, expressionEvaluator);
-
-		PersistentEntityParameterValueProvider<CassandraPersistentProperty> parameterValueProvider =
-				getParameterValueProvider(entity, rowValueProvider);
-
-		EntityInstantiator instantiator = instantiators.getInstantiatorFor(entity);
-
-		S instance = instantiator.createInstance(entity, parameterValueProvider);
-
-		readPropertiesFromRow(entity, rowValueProvider, getConvertingAccessor(instance, entity));
-
-		return instance;
+		return doRead(entity, row, expressionEvaluator -> new BasicCassandraRowValueProvider(row, expressionEvaluator));
 	}
 
 	protected <S> S readEntityFromUdt(CassandraPersistentEntity<S> entity, UDTValue udtValue) {
+		return doRead(entity, udtValue,
+				expressionEvaluator -> new CassandraUDTValueProvider(udtValue, getCodecRegistry(), expressionEvaluator));
+	}
 
-		DefaultSpELExpressionEvaluator expressionEvaluator = new DefaultSpELExpressionEvaluator(udtValue, spELContext);
+	protected <S> S readEntityFromTuple(CassandraPersistentEntity<S> entity, TupleValue tupleValue) {
+		return doRead(entity, tupleValue,
+				expressionEvaluator -> new CassandraTupleValueProvider(tupleValue, getCodecRegistry(), expressionEvaluator));
+	}
 
-		CassandraUDTValueProvider valueProvider =
-				new CassandraUDTValueProvider(udtValue, CodecRegistry.DEFAULT_INSTANCE, expressionEvaluator);
+	protected <S, V> S doRead(CassandraPersistentEntity<S> entity, V value,
+			Function<SpELExpressionEvaluator, CassandraValueProvider> valueProviderSupplier) {
 
-		PersistentEntityParameterValueProvider<CassandraPersistentProperty> parameterValueProvider =
-				getParameterValueProvider(entity, valueProvider);
+		DefaultSpELExpressionEvaluator expressionEvaluator = new DefaultSpELExpressionEvaluator(value, spELContext);
+
+		CassandraValueProvider valueProvider = valueProviderSupplier.apply(expressionEvaluator);
+
+		PersistentEntityParameterValueProvider<CassandraPersistentProperty> parameterValueProvider = getParameterValueProvider(
+				entity, valueProvider);
 
 		EntityInstantiator instantiator = instantiators.getInstantiatorFor(entity);
 
@@ -221,13 +220,12 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	private <S> PersistentEntityParameterValueProvider<CassandraPersistentProperty> getParameterValueProvider(
 			CassandraPersistentEntity<S> entity, CassandraValueProvider valueProvider) {
 
-		return new PersistentEntityParameterValueProvider<>(entity,
-				new MappingAndConvertingValueProvider(valueProvider), null);
+		return new PersistentEntityParameterValueProvider<>(entity, new MappingAndConvertingValueProvider(valueProvider),
+				null);
 	}
 
 	protected void readPropertiesFromRow(CassandraPersistentEntity<?> entity, CassandraRowValueProvider row,
 			PersistentPropertyAccessor propertyAccessor) {
-
 		readProperties(entity, row, propertyAccessor);
 	}
 
@@ -331,8 +329,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 		try {
 			return (Class<T>) ClassUtils.forName(entity.getName(), this.beanClassLoader);
-		}
-		catch (ClassNotFoundException | LinkageError ignore) {
+		} catch (ClassNotFoundException | LinkageError ignore) {
 			return entity;
 		}
 	}
@@ -358,7 +355,9 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		} else if (sink instanceof Delete.Where) {
 			writeDeleteWhereFromObject(source, (Delete.Where) sink, entity);
 		} else if (sink instanceof UDTValue) {
-			writeUDTValueWhereFromObject(getConvertingAccessor(source, entity), (UDTValue) sink, entity);
+			writeUDTValue(getConvertingAccessor(source, entity), (UDTValue) sink, entity);
+		} else if (sink instanceof TupleValue) {
+			writeTupleValue(getConvertingAccessor(source, entity), (TupleValue) sink, entity);
 		} else {
 			throw new MappingException("Unknown write target " + sink.getClass().getName());
 		}
@@ -389,8 +388,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 					continue;
 				}
 
-				CassandraPersistentEntity<?> compositePrimaryKey =
-					getMappingContext().getRequiredPersistentEntity(property);
+				CassandraPersistentEntity<?> compositePrimaryKey = getMappingContext().getRequiredPersistentEntity(property);
 
 				writeInsertFromWrapper(getConvertingAccessor(value, compositePrimaryKey), insert, compositePrimaryKey);
 
@@ -402,10 +400,10 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			}
 
 			if (log.isDebugEnabled()) {
-				log.debug("Adding insert.value [{}] - [{}]", property.getColumnName().toCql(), value);
+				log.debug("Adding insert.value [{}] - [{}]", property.getRequiredColumnName().toCql(), value);
 			}
 
-			insert.value(property.getColumnName().toCql(), value);
+			insert.value(property.getRequiredColumnName().toCql(), value);
 		}
 	}
 
@@ -430,8 +428,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 					continue;
 				}
 
-				CassandraPersistentEntity<?> compositePrimaryKey =
-						getMappingContext().getRequiredPersistentEntity(property);
+				CassandraPersistentEntity<?> compositePrimaryKey = getMappingContext().getRequiredPersistentEntity(property);
 
 				writeMapFromWrapper(getConvertingAccessor(value, compositePrimaryKey), insert, compositePrimaryKey);
 
@@ -439,10 +436,10 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			}
 
 			if (log.isDebugEnabled()) {
-				log.debug("Adding map.entry [{}] - [{}]", property.getColumnName().toCql(), value);
+				log.debug("Adding map.entry [{}] - [{}]", property.getRequiredColumnName().toCql(), value);
 			}
 
-			insert.put(property.getColumnName().toCql(), value);
+			insert.put(property.getRequiredColumnName().toCql(), value);
 		}
 	}
 
@@ -459,8 +456,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 			if (property.isCompositePrimaryKey()) {
 
-				CassandraPersistentEntity<?> compositePrimaryKey =
-						getMappingContext().getRequiredPersistentEntity(property);
+				CassandraPersistentEntity<?> compositePrimaryKey = getMappingContext().getRequiredPersistentEntity(property);
 
 				if (value == null) {
 					continue;
@@ -472,9 +468,9 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			}
 
 			if (isPrimaryKeyPart(property)) {
-				update.where(QueryBuilder.eq(property.getColumnName().toCql(), value));
+				update.where(QueryBuilder.eq(property.getRequiredColumnName().toCql(), value));
 			} else {
-				update.with(QueryBuilder.set(property.getColumnName().toCql(), value));
+				update.with(QueryBuilder.set(property.getRequiredColumnName().toCql(), value));
 			}
 		}
 	}
@@ -487,7 +483,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		getWhereClauses(object, entity).forEach(where::and);
 	}
 
-	protected void writeUDTValueWhereFromObject(ConvertingPropertyAccessor accessor, UDTValue udtValue,
+	protected void writeUDTValue(ConvertingPropertyAccessor accessor, UDTValue udtValue,
 			CassandraPersistentEntity<?> entity) {
 
 		for (CassandraPersistentProperty property : entity) {
@@ -495,16 +491,38 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			Object value = getWriteValue(property, accessor);
 
 			if (log.isDebugEnabled()) {
-				log.debug("writeUDTValueWhereFromObject Property.type {}, Property.value {}", property.getType().getName(), value);
+				log.debug("writeUDTValueWhereFromObject Property.type {}, Property.value {}", property.getType().getName(),
+						value);
 			}
 
 			if (log.isDebugEnabled()) {
-				log.debug("Adding udt.value [{}] - [{}]", property.getColumnName().toCql(), value);
+				log.debug("Adding udt.value [{}] - [{}]", property.getRequiredColumnName().toCql(), value);
 			}
 
-			TypeCodec<Object> typeCodec = CodecRegistry.DEFAULT_INSTANCE.codecFor(getMappingContext().getDataType(property));
+			TypeCodec<Object> typeCodec = getCodecRegistry().codecFor(getMappingContext().getDataType(property));
 
-			udtValue.set(property.getColumnName().toCql(), value, typeCodec);
+			udtValue.set(property.getRequiredColumnName().toCql(), value, typeCodec);
+		}
+	}
+
+	protected void writeTupleValue(ConvertingPropertyAccessor accessor, TupleValue tupleValue,
+			CassandraPersistentEntity<?> entity) {
+
+		for (CassandraPersistentProperty property : entity) {
+
+			Object value = getWriteValue(property, accessor);
+
+			if (log.isDebugEnabled()) {
+				log.debug("writeTupleValue Property.type {}, Property.value {}", property.getType().getName(), value);
+			}
+
+			if (log.isDebugEnabled()) {
+				log.debug("Adding tuple value [{}] - [{}]", property.getOrdinal(), value);
+			}
+
+			TypeCodec<Object> typeCodec = getCodecRegistry().codecFor(mappingContext.getDataType(property));
+
+			tupleValue.set(property.getRequiredOrdinal(), value, typeCodec);
 		}
 	}
 
@@ -544,16 +562,16 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 						String.format("Cannot use [%s] as composite Id for [%s]", id, entity.getName()));
 			}
 
-			CassandraPersistentEntity<?> compositePrimaryKey =
-					getMappingContext().getRequiredPersistentEntity(compositeIdProperty);
+			CassandraPersistentEntity<?> compositePrimaryKey = getMappingContext()
+					.getRequiredPersistentEntity(compositeIdProperty);
 
 			return getWhereClauses(getConvertingAccessor(id, compositePrimaryKey), compositePrimaryKey);
 		}
 
 		Class<?> targetType = getTargetType(idProperty);
 
-		return Collections.singleton(
-				QueryBuilder.eq(idProperty.getColumnName().toCql(), getPotentiallyConvertedSimpleValue(id, targetType)));
+		return Collections.singleton(QueryBuilder.eq(idProperty.getRequiredColumnName().toCql(),
+				getPotentiallyConvertedSimpleValue(id, targetType)));
 	}
 
 	@Nullable
@@ -570,8 +588,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		return source;
 	}
 
-	private Collection<Clause> getWhereClauses(ConvertingPropertyAccessor accessor,
-			CassandraPersistentEntity<?> entity) {
+	private Collection<Clause> getWhereClauses(ConvertingPropertyAccessor accessor, CassandraPersistentEntity<?> entity) {
 
 		Assert.isTrue(entity.isCompositePrimaryKey(),
 				String.format("Entity [%s] is not a composite primary key", entity.getName()));
@@ -581,7 +598,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		for (CassandraPersistentProperty property : entity) {
 			TypeCodec<Object> codec = getCodec(property);
 			Object value = accessor.getProperty(property, codec.getJavaType().getRawType());
-			clauses.add(QueryBuilder.eq(property.getColumnName().toCql(), value));
+			clauses.add(QueryBuilder.eq(property.getRequiredColumnName().toCql(), value));
 		}
 
 		return clauses;
@@ -599,13 +616,12 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 			if (persistentProperty == null) {
 				throw new IllegalArgumentException(String.format(
-						"MapId contains references [%s] that is an unknown property of [%s]",
-								entry.getKey(), entity.getName()));
+						"MapId contains references [%s] that is an unknown property of [%s]", entry.getKey(), entity.getName()));
 			}
 
 			Object writeValue = getWriteValue(entry.getValue(), persistentProperty.getTypeInformation());
 
-			clauses.add(QueryBuilder.eq(persistentProperty.getColumnName().toCql(), writeValue));
+			clauses.add(QueryBuilder.eq(persistentProperty.getRequiredColumnName().toCql(), writeValue));
 		}
 
 		return clauses;
@@ -701,11 +717,11 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 		DataType dataType = getMappingContext().getDataType(property);
 
-		if (dataType instanceof UserType) {
+		if (dataType instanceof UserType || dataType instanceof TupleType) {
 			return property.getType();
 		}
 
-		TypeCodec<Object> codec = CodecRegistry.DEFAULT_INSTANCE.codecFor(getMappingContext().getDataType(property));
+		TypeCodec<Object> codec = getCodecRegistry().codecFor(getMappingContext().getDataType(property));
 
 		return codec.getJavaType().getRawType();
 	}
@@ -745,8 +761,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 		if (getCustomConversions().hasCustomWriteTarget(value.getClass(), requestedTargetType)) {
 
-			Class<?> resolvedTargetType = getCustomConversions()
-					.getCustomWriteTarget(value.getClass(), requestedTargetType)
+			Class<?> resolvedTargetType = getCustomConversions().getCustomWriteTarget(value.getClass(), requestedTargetType)
 					.orElse(requestedTargetType);
 
 			return getConversionService().convert(value, resolvedTargetType);
@@ -754,11 +769,9 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 		if (getCustomConversions().hasCustomWriteTarget(value.getClass())) {
 
-			Class<?> resolvedTargetType = getCustomConversions()
-					.getCustomWriteTarget(value.getClass())
-					.orElseThrow(() -> new IllegalStateException(
-							String.format("Unable to determined custom write target for value type [%s]",
-									value.getClass().getName())));
+			Class<?> resolvedTargetType = getCustomConversions().getCustomWriteTarget(value.getClass())
+					.orElseThrow(() -> new IllegalStateException(String
+							.format("Unable to determined custom write target for value type [%s]", value.getClass().getName())));
 
 			return getConversionService().convert(value, resolvedTargetType);
 		}
@@ -768,7 +781,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		}
 
 		TypeInformation<?> type = typeInformation != null ? typeInformation
-			: ClassTypeInformation.from((Class) value.getClass());
+				: ClassTypeInformation.from((Class) value.getClass());
 
 		if (value instanceof Collection) {
 			return writeCollectionInternal((Collection<Object>) value, type);
@@ -782,13 +795,25 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 		BasicCassandraPersistentEntity<?> entity = getMappingContext().getPersistentEntity(actualType.getType());
 
-		if (entity != null && entity.isUserDefinedType()) {
+		if (entity != null) {
 
-			UDTValue udtValue = entity.getUserType().newValue();
+			if (entity.isUserDefinedType()) {
 
-			write(value, udtValue, entity);
+				UDTValue udtValue = entity.getUserType().newValue();
 
-			return udtValue;
+				write(value, udtValue, entity);
+
+				return udtValue;
+			}
+
+			if (entity.isTupleType()) {
+
+				TupleValue tupleValue = mappingContext.getTupleType(entity).newValue();
+
+				write(value, tupleValue, entity);
+
+				return tupleValue;
+			}
 		}
 
 		return value;
@@ -815,8 +840,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		TypeInformation<?> valueType = type.getRequiredMapValueType();
 
 		for (Entry<Object, Object> entry : source.entrySet()) {
-			converted.put(convertToColumnType(entry.getKey(), keyType),
-				convertToColumnType(entry.getValue(), valueType));
+			converted.put(convertToColumnType(entry.getKey(), keyType), convertToColumnType(entry.getValue(), valueType));
 		}
 
 		return converted;
@@ -935,8 +959,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 				Collection<Object> original = (Collection<Object>) value;
 
-				Collection<Object> converted =
-						CollectionFactory.createCollection(typeInformation.getType(), original.size());
+				Collection<Object> converted = CollectionFactory.createCollection(typeInformation.getType(), original.size());
 
 				for (Object element : original) {
 					converted.add(getConversionService().convert(element, typeInformation.getRequiredActualType().getType()));
@@ -954,11 +977,24 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			return readMapInternal((Map<Object, Object>) value, typeInformation);
 		}
 
-		BasicCassandraPersistentEntity<?> persistentEntity =
-				getMappingContext().getPersistentEntity(typeInformation.getRequiredActualType());
+		if (value instanceof UDTValue) {
 
-		if (persistentEntity != null && persistentEntity.isUserDefinedType() && value instanceof UDTValue) {
-			return readEntityFromUdt(persistentEntity, (UDTValue) value);
+			BasicCassandraPersistentEntity<?> udtEntity = getMappingContext()
+					.getPersistentEntity(typeInformation.getRequiredActualType());
+
+			if (udtEntity != null && udtEntity.isUserDefinedType()) {
+				return readEntityFromUdt(udtEntity, (UDTValue) value);
+			}
+		}
+
+		if (value instanceof TupleValue) {
+
+			BasicCassandraPersistentEntity<?> tupleEntity = getMappingContext()
+					.getPersistentEntity(typeInformation.getRequiredActualType());
+
+			if (tupleEntity != null) {
+				return readEntityFromTuple(tupleEntity, (TupleValue) value);
+			}
 		}
 
 		return getPotentiallyConvertedSimpleRead(value, typeInformation.getType());
@@ -989,11 +1025,17 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 		BasicCassandraPersistentEntity<?> entity = getMappingContext().getPersistentEntity(elementType);
 
-		if (entity != null && entity.isUserDefinedType()) {
-			for (Object udtValue : source) {
-				collection.add(readEntityFromUdt(entity, (UDTValue) udtValue));
-			}
+		if (entity != null) {
 
+			if (entity.isUserDefinedType()) {
+				for (Object udtValue : source) {
+					collection.add(readEntityFromUdt(entity, (UDTValue) udtValue));
+				}
+			} else if (entity.isTupleType()) {
+				for (Object tupleValue : source) {
+					collection.add(readEntityFromTuple(entity, (TupleValue) tupleValue));
+				}
+			}
 		} else {
 			for (Object element : source) {
 				collection.add(getPotentiallyConvertedSimpleRead(element, elementType));
@@ -1064,7 +1106,11 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	}
 
 	private TypeCodec<Object> getCodec(CassandraPersistentProperty property) {
-		return CodecRegistry.DEFAULT_INSTANCE.codecFor(mappingContext.getDataType(property));
+		return getCodecRegistry().codecFor(mappingContext.getDataType(property));
+	}
+
+	private static CodecRegistry getCodecRegistry() {
+		return CodecRegistry.DEFAULT_INSTANCE;
 	}
 
 	/**

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/QueryMapper.java

@@ -125,6 +125,11 @@ public class QueryMapper {
 						"Cannot use composite primary key directly. Reference a property of the composite primary key");
 			});
 
+			field.getProperty().filter(it -> it.getOrdinal() != null).ifPresent(it -> {
+				throw new IllegalArgumentException(
+						String.format("Cannot reference tuple value elements, property [%s]", field.getMappedKey()));
+			});
+
 			Object value = predicate.getValue();
 			TypeInformation<?> typeInformation = getTypeInformation(field, value);
 			Object mappedValue = value != null ? getConverter().convertToColumnType(value, typeInformation) : null;
@@ -182,7 +187,7 @@ public class QueryMapper {
 				CassandraPersistentEntity<?> primaryKeyEntity = mappingContext.getRequiredPersistentEntity(property);
 				addColumns(primaryKeyEntity, selectors);
 			} else {
-				selectors.add(ColumnSelector.from(property.getColumnName().toCql()));
+				selectors.add(ColumnSelector.from(property.getRequiredColumnName().toCql()));
 			}
 		});
 	}
@@ -265,7 +270,7 @@ public class QueryMapper {
 				}
 
 				if (seen.add(property)) {
-					columnNames.add(property.getColumnName().toCql());
+					columnNames.add(property.getRequiredColumnName().toCql());
 				}
 			});
 		}
@@ -309,7 +314,7 @@ public class QueryMapper {
 						throw new IllegalArgumentException(
 								"Cannot use composite primary key directly. Reference a property of the composite primary key");
 					}
-					return cassandraPersistentProperty.getColumnName();
+					return cassandraPersistentProperty.getRequiredColumnName();
 				});
 			}
 

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/convert/UpdateMapper.java

@@ -85,6 +85,11 @@ public class UpdateMapper extends QueryMapper {
 
 			Field field = createPropertyField(entity, assignmentOp.getColumnName());
 
+			field.getProperty().filter(it -> it.getOrdinal() != null).ifPresent(it -> {
+				throw new IllegalArgumentException(
+						String.format("Cannot reference tuple value elements, property [%s]", field.getMappedKey()));
+			});
+
 			mapped.add(getMappedUpdateOperation(assignmentOp, field));
 		}
 
@@ -127,8 +132,8 @@ public class UpdateMapper extends QueryMapper {
 			Assert.state(op.getValue() != null,
 					() -> String.format("SetAtKeyOp for %s attempts to set null", field.getProperty()));
 
-			Optional<? extends TypeInformation<?>> typeInformation =
-					field.getProperty().map(PersistentProperty::getTypeInformation);
+			Optional<? extends TypeInformation<?>> typeInformation = field.getProperty()
+					.map(PersistentProperty::getTypeInformation);
 
 			Optional<TypeInformation<?>> keyType = typeInformation.map(TypeInformation::getComponentType);
 			Optional<TypeInformation<?>> valueType = typeInformation.map(TypeInformation::getMapValueType);
@@ -162,8 +167,8 @@ public class UpdateMapper extends QueryMapper {
 
 			if (collection.isEmpty()) {
 
-				DataType.Name dataType = field.getProperty().map(property ->
-						getMappingContext().getDataType(property)).map(DataType::getName).orElse(Name.LIST);
+				DataType.Name dataType = field.getProperty().map(property -> getMappingContext().getDataType(property))
+						.map(DataType::getName).orElse(Name.LIST);
 
 				if (dataType == Name.SET) {
 					return new SetOp(field.getMappedKey(), Collections.emptySet());

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/BasicCassandraPersistentEntity.java

@@ -17,6 +17,7 @@ package org.springframework.data.cassandra.core.mapping;
 
 import static org.springframework.data.cassandra.core.cql.CqlIdentifier.*;
 
+import java.util.Comparator;
 import java.util.Optional;
 
 import org.springframework.beans.BeansException;
@@ -36,6 +37,7 @@ import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
+import com.datastax.driver.core.TupleType;
 import com.datastax.driver.core.UserType;
 
 /**
@@ -83,6 +85,23 @@ public class BasicCassandraPersistentEntity<T> extends BasicPersistentEntity<T,
 		setVerifier(verifier);
 	}
 
+	/**
+	 * Create a new {@link BasicCassandraPersistentEntity} with the given {@link TypeInformation}. Will default the table
+	 * name to the entity's simple type name.
+	 *
+	 * @param typeInformation must not be {@literal null}.
+	 * @param verifier must not be {@literal null}.
+	 * @param comparator must not be {@literal null}.
+	 * @since 2.1
+	 */
+	protected BasicCassandraPersistentEntity(TypeInformation<T> typeInformation,
+			CassandraPersistentEntityMetadataVerifier verifier, Comparator<CassandraPersistentProperty> comparator) {
+
+		super(typeInformation, comparator);
+
+		setVerifier(verifier);
+	}
+
 	protected CqlIdentifier determineTableName() {
 
 		Table annotation = findAnnotation(Table.class);
@@ -223,4 +242,21 @@ public class BasicCassandraPersistentEntity<T> extends BasicPersistentEntity<T,
 	public UserType getUserType() {
 		return null;
 	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity#isTupleType()
+	 */
+	@Override
+	public boolean isTupleType() {
+		return false;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity#getTupleType()
+	 */
+	@Override
+	@Nullable
+	public TupleType getTupleType() {
+		return null;
+	}
 }

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/BasicCassandraPersistentProperty.java

@@ -146,6 +146,15 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 		return this.columnName;
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty#getOrdinal()
+	 */
+	@Nullable
+	@Override
+	public Integer getOrdinal() {
+		return null;
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty#getPrimaryKeyOrdering()
 	 */
@@ -397,7 +406,7 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 
 		if (changed) {
 
-			CqlIdentifier columnName = getColumnName();
+			CqlIdentifier columnName = getRequiredColumnName();
 			setColumnName(of(columnName.getUnquoted(), forceQuote));
 		}
 	}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/BasicCassandraPersistentTupleEntity.java

@@ -0,0 +1,113 @@
+/*
+ * Copyright 2018 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.cassandra.core.mapping;
+
+import java.util.Comparator;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+import org.springframework.data.mapping.MappingException;
+import org.springframework.data.util.Lazy;
+import org.springframework.data.util.TypeInformation;
+import org.springframework.util.Assert;
+
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.TupleType;
+
+/**
+ * Cassandra Tuple-specific {@link org.springframework.data.mapping.PersistentEntity} for a mapped tuples. Mapped tuples
+ * are nested level entities that can be referred from a {@link CassandraPersistentEntity}.
+ *
+ * @author Mark Paluch
+ * @since 2.1
+ * @see Tuple
+ * @see Element
+ */
+public class BasicCassandraPersistentTupleEntity<T> extends BasicCassandraPersistentEntity<T> {
+
+	private final Lazy<TupleType> tupleType;
+
+	/**
+	 * Creates a new {@link BasicCassandraPersistentTupleEntity} given {@link TypeInformation} and
+	 * {@link TupleTypeFactory}.
+	 *
+	 * @param information must not be {@literal null}.
+	 * @param tupleTypeFactory must not be {@literal null}.
+	 */
+	public BasicCassandraPersistentTupleEntity(TypeInformation<T> information, TupleTypeFactory tupleTypeFactory) {
+
+		super(information, CassandraPersistentTupleMetadataVerifier.INSTANCE, TuplePropertyComparator.INSTANCE);
+
+		Assert.notNull(tupleTypeFactory, "TupleTypeFactory must not be null");
+
+		this.tupleType = Lazy.of(() -> tupleTypeFactory.create(getTupleFieldTypes()));
+	}
+
+	private List<DataType> getTupleFieldTypes() {
+
+		return StreamSupport.stream(spliterator(), false) //
+				.sorted(TuplePropertyComparator.INSTANCE) //
+				.map(CassandraPersistentProperty::getDataType) //
+				.collect(Collectors.toList());
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mapping.model.BasicPersistentEntity#verify()
+	 */
+	@Override
+	public void verify() throws MappingException {
+
+		super.verify();
+
+		CassandraPersistentTupleMetadataVerifier.INSTANCE.verify(this);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentEntity#isTupleType()
+	 */
+	@Override
+	public boolean isTupleType() {
+		return true;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentEntity#getTupleType()
+	 */
+	@Override
+	public TupleType getTupleType() {
+		return tupleType.get();
+	}
+
+	/**
+	 * {@link CassandraPersistentProperty} comparator using to sort properties by their
+	 * {@link CassandraPersistentProperty#getRequiredOrdinal()}.
+	 *
+	 * @see Element
+	 */
+	enum TuplePropertyComparator implements Comparator<CassandraPersistentProperty> {
+
+		INSTANCE;
+
+		/* (non-Javadoc)
+		 * @see java.util.Comparator#compare(java.lang.Object, java.lang.Object)
+		 */
+		@Override
+		public int compare(CassandraPersistentProperty o1, CassandraPersistentProperty o2) {
+			return Integer.compare(o1.getRequiredOrdinal(), o2.getRequiredOrdinal());
+		}
+	}
+}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/BasicCassandraPersistentTupleProperty.java

@@ -0,0 +1,145 @@
+/*
+ * Copyright 2018 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.cassandra.core.mapping;
+
+import org.springframework.data.cassandra.core.cql.CqlIdentifier;
+import org.springframework.data.mapping.MappingException;
+import org.springframework.data.mapping.model.Property;
+import org.springframework.data.mapping.model.SimpleTypeHolder;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Cassandra Tuple specific {@link CassandraPersistentProperty} implementation.
+ *
+ * @author Mark Paluch
+ * @since 2.1
+ * @see Element
+ */
+public class BasicCassandraPersistentTupleProperty extends BasicCassandraPersistentProperty {
+
+	private final @Nullable Integer ordinal;
+
+	/**
+	 * Create a new {@link BasicCassandraPersistentTupleProperty}.
+	 *
+	 * @param property the actual {@link Property} in the domain entity corresponding to this persistent entity.
+	 * @param owner the containing object or {@link CassandraPersistentEntity} of this persistent property.
+	 * @param simpleTypeHolder mapping of Java [simple|wrapper] types to Cassandra data types.
+	 */
+	public BasicCassandraPersistentTupleProperty(Property property, CassandraPersistentEntity<?> owner,
+			SimpleTypeHolder simpleTypeHolder) {
+		this(property, owner, simpleTypeHolder, null);
+	}
+
+	/**
+	 * Create a new {@link BasicCassandraPersistentTupleProperty}.
+	 *
+	 * @param property the actual {@link Property} in the domain entity corresponding to this persistent entity.
+	 * @param owner the containing object or {@link CassandraPersistentEntity} of this persistent property.
+	 * @param simpleTypeHolder mapping of Java [simple|wrapper] types to Cassandra data types.
+	 * @param userTypeResolver resolver for user-defined types.
+	 */
+	public BasicCassandraPersistentTupleProperty(Property property, CassandraPersistentEntity<?> owner,
+			SimpleTypeHolder simpleTypeHolder, @Nullable UserTypeResolver userTypeResolver) {
+
+		super(property, owner, simpleTypeHolder, userTypeResolver);
+
+		this.ordinal = findOrdinal();
+	}
+
+	@Nullable
+	private Integer findOrdinal() {
+
+		if (isTransient()) {
+			return null;
+		}
+
+		int ordinal;
+
+		try {
+			ordinal = getRequiredAnnotation(Element.class).value();
+		} catch (IllegalStateException e) {
+			throw new MappingException(
+					String.format("Missing @Element annotation in mapped tuple type for property [%s] in entity [%s]", getName(),
+							getOwner().getName()),
+					e);
+		}
+
+		Assert.isTrue(ordinal >= 0,
+				String.format("Element ordinal must be greater or equal to zero for property [%s] in entity [%s]", getName(),
+						getOwner().getName()));
+
+		return ordinal;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty#getColumnName()
+	 */
+	@Override
+	public CqlIdentifier getColumnName() {
+		return null;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentProperty#getOrdinal()
+	 */
+	@Nullable
+	@Override
+	public Integer getOrdinal() {
+		return ordinal;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty#isCompositePrimaryKey()
+	 */
+	@Override
+	public boolean isCompositePrimaryKey() {
+		return false;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty#isPrimaryKeyColumn()
+	 */
+	@Override
+	public boolean isPrimaryKeyColumn() {
+		return false;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty#isPartitionKeyColumn()
+	 */
+	@Override
+	public boolean isPartitionKeyColumn() {
+		return false;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty#isClusterKeyColumn()
+	 */
+	@Override
+	public boolean isClusterKeyColumn() {
+		return false;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty#setColumnName(org.springframework.data.cassandra.core.cql.CqlIdentifier)
+	 */
+	@Override
+	public void setColumnName(CqlIdentifier columnName) {
+		throw new UnsupportedOperationException("Cannot set a column name on a property representing a tuple element");
+	}
+}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContext.java

@@ -15,20 +15,10 @@
  */
 package org.springframework.data.cassandra.core.mapping;
 
-import static org.springframework.data.cassandra.core.cql.keyspace.CreateTableSpecification.createTable;
-import static org.springframework.data.cassandra.core.mapping.CassandraSimpleTypeHolder.getDataTypeFor;
+import static org.springframework.data.cassandra.core.cql.keyspace.CreateTableSpecification.*;
+import static org.springframework.data.cassandra.core.mapping.CassandraSimpleTypeHolder.*;
 
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.Set;
+import java.util.*;
 import java.util.function.Supplier;
 import java.util.stream.StreamSupport;
 
@@ -57,10 +47,8 @@ import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.StringUtils;
 
-import com.datastax.driver.core.CodecRegistry;
 import com.datastax.driver.core.DataType;
 import com.datastax.driver.core.DataType.Name;
-import com.datastax.driver.core.ProtocolVersion;
 import com.datastax.driver.core.TupleType;
 
 /**
@@ -81,16 +69,17 @@ public class CassandraMappingContext
 
 	private @Nullable ClassLoader beanClassLoader;
 
-	private CassandraPersistentEntityMetadataVerifier verifier =
-			new CompositeCassandraPersistentEntityMetadataVerifier();
+	private CassandraPersistentEntityMetadataVerifier verifier = new CompositeCassandraPersistentEntityMetadataVerifier();
 
-	private CustomConversions customConversions =
-			new CustomConversions(StoreConversions.of(CassandraSimpleTypeHolder.HOLDER), Collections.emptyList());
+	private CustomConversions customConversions = new CustomConversions(
+			StoreConversions.of(CassandraSimpleTypeHolder.HOLDER), Collections.emptyList());
 
 	private Mapping mapping = new Mapping();
 
 	private @Nullable UserTypeResolver userTypeResolver;
 
+	private TupleTypeFactory tupleTypeFactory = CodecRegistryTupleTypeFactory.DEFAULT;
+
 	// caches
 	private final Map<CqlIdentifier, Set<CassandraPersistentEntity<?>>> entitySetsByTableName = new HashMap<>();
 
@@ -108,6 +97,27 @@ public class CassandraMappingContext
 		setSimpleTypeHolder(CassandraSimpleTypeHolder.HOLDER);
 	}
 
+	/**
+	 * Create a new {@link CassandraMappingContext} given {@link UserTypeResolver} and {@link TupleTypeFactory}.
+	 *
+	 * @param userTypeResolver must not be {@literal null}.
+	 * @param tupleTypeFactory must not be {@literal null}.
+	 * @since 2.1
+	 */
+	public CassandraMappingContext(UserTypeResolver userTypeResolver, TupleTypeFactory tupleTypeFactory) {
+
+		Assert.notNull(userTypeResolver, "UserTypeResolver must not be null");
+		Assert.notNull(tupleTypeFactory, "TupleTypeFactory must not be null");
+
+		StoreConversions storeConversions = StoreConversions.of(CassandraSimpleTypeHolder.HOLDER);
+
+		setUserTypeResolver(userTypeResolver);
+		setTupleTypeFactory(tupleTypeFactory);
+
+		setCustomConversions(new CustomConversions(storeConversions, Collections.emptyList()));
+		setSimpleTypeHolder(CassandraSimpleTypeHolder.HOLDER);
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.mapping.context.AbstractMappingContext#initialize()
 	 */
@@ -141,10 +151,9 @@ public class CassandraMappingContext
 
 		try {
 			return ClassUtils.forName(entityClassName, this.beanClassLoader);
-		}
-		catch (ClassNotFoundException cause) {
-			throw new IllegalStateException(
-					String.format("Unknown persistent entity type name [%s]", entityClassName), cause);
+		} catch (ClassNotFoundException cause) {
+			throw new IllegalStateException(String.format("Unknown persistent entity type name [%s]", entityClassName),
+					cause);
 		}
 	}
 
@@ -237,6 +246,19 @@ public class CassandraMappingContext
 		this.userTypeResolver = userTypeResolver;
 	}
 
+	/**
+	 * Sets the {@link TupleTypeFactory}.
+	 *
+	 * @param tupleTypeFactory must not be {@literal null}.
+	 * @since 2.1
+	 */
+	public void setTupleTypeFactory(TupleTypeFactory tupleTypeFactory) {
+
+		Assert.notNull(tupleTypeFactory, "TupleTypeFactory must not be null");
+
+		this.tupleTypeFactory = tupleTypeFactory;
+	}
+
 	/**
 	 * @param verifier The verifier to set.
 	 */
@@ -269,12 +291,12 @@ public class CassandraMappingContext
 			}
 			// now do some caching of the entity
 
-			Set<CassandraPersistentEntity<?>> entities = this.entitySetsByTableName
-					.computeIfAbsent(entity.getTableName(), cqlIdentifier -> new HashSet<>());
+			Set<CassandraPersistentEntity<?>> entities = this.entitySetsByTableName.computeIfAbsent(entity.getTableName(),
+					cqlIdentifier -> new HashSet<>());
 
 			entities.add(entity);
 
-			if (!entity.isUserDefinedType() && entity.isAnnotationPresent(Table.class)) {
+			if (!entity.isUserDefinedType() && !entity.isTupleType() && entity.isAnnotationPresent(Table.class)) {
 				this.tableEntities.add(entity);
 			}
 
@@ -299,10 +321,17 @@ public class CassandraMappingContext
 	protected <T> BasicCassandraPersistentEntity<T> createPersistentEntity(TypeInformation<T> typeInformation) {
 
 		BasicCassandraPersistentEntity<T> entity = Optional.ofNullable(resolveUserDefinedType(typeInformation))
-				.<BasicCassandraPersistentEntity<T>>map(resolvedUserDefinedType ->
-					new CassandraUserTypePersistentEntity<>(typeInformation, getVerifier(), resolveUserTypeResolver()))
-				.orElseGet(() ->
-					new BasicCassandraPersistentEntity<>(typeInformation, getVerifier()));
+				.<BasicCassandraPersistentEntity<T>> map(resolvedUserDefinedType -> new CassandraUserTypePersistentEntity<>(
+						typeInformation, getVerifier(), resolveUserTypeResolver()))
+				.orElseGet(() -> {
+					boolean tuple = AnnotatedElementUtils.hasAnnotation(typeInformation.getType(), Tuple.class);
+
+					if (tuple) {
+						return new BasicCassandraPersistentTupleEntity<>(typeInformation, tupleTypeFactory);
+					}
+
+					return new BasicCassandraPersistentEntity<>(typeInformation, getVerifier());
+				});
 
 		Optional.ofNullable(this.applicationContext).ifPresent(entity::setApplicationContext);
 
@@ -331,8 +360,15 @@ public class CassandraMappingContext
 	protected CassandraPersistentProperty createPersistentProperty(Property property,
 			BasicCassandraPersistentEntity<?> owner, SimpleTypeHolder simpleTypeHolder) {
 
-		BasicCassandraPersistentProperty persistentProperty =
-				new BasicCassandraPersistentProperty(property, owner, simpleTypeHolder, this.userTypeResolver);
+		BasicCassandraPersistentProperty persistentProperty;
+
+		if (owner.isTupleType()) {
+			persistentProperty = new BasicCassandraPersistentTupleProperty(property, owner, simpleTypeHolder,
+					this.userTypeResolver);
+		} else {
+			persistentProperty = new BasicCassandraPersistentProperty(property, owner, simpleTypeHolder,
+					this.userTypeResolver);
+		}
 
 		Optional.ofNullable(this.applicationContext).ifPresent(persistentProperty::setApplicationContext);
 
@@ -372,12 +408,9 @@ public class CassandraMappingContext
 
 	private boolean hasReferencedUserType(CqlIdentifier identifier) {
 
-		return getPersistentEntities().stream()
-				.flatMap(entity -> StreamSupport.stream(entity.spliterator(), false))
+		return getPersistentEntities().stream().flatMap(entity -> StreamSupport.stream(entity.spliterator(), false))
 				.flatMap(it -> Optionals.toStream(Optional.ofNullable(it.findAnnotation(CassandraType.class))))
-				.map(CassandraType::userTypeName)
-				.filter(StringUtils::hasText)
-				.map(CqlIdentifier::of)
+				.map(CassandraType::userTypeName).filter(StringUtils::hasText).map(CqlIdentifier::of)
 				.anyMatch(identifier::equals);
 	}
 
@@ -400,24 +433,22 @@ public class CassandraMappingContext
 
 				for (CassandraPersistentProperty primaryKeyProperty : primaryKeyEntity) {
 					if (primaryKeyProperty.isPartitionKeyColumn()) {
-						specification.partitionKeyColumn(primaryKeyProperty.getColumnName(),
+						specification.partitionKeyColumn(primaryKeyProperty.getRequiredColumnName(),
 								getDataType(primaryKeyProperty));
 					} else { // cluster column
-						specification.clusteredKeyColumn(primaryKeyProperty.getColumnName(),
+						specification.clusteredKeyColumn(primaryKeyProperty.getRequiredColumnName(),
 								getDataType(primaryKeyProperty), primaryKeyProperty.getPrimaryKeyOrdering());
 					}
 				}
-			}
-			else {
+			} else {
 				if (property.isIdProperty() || property.isPartitionKeyColumn()) {
-					specification.partitionKeyColumn(property.getColumnName(),
+					specification.partitionKeyColumn(property.getRequiredColumnName(),
 							UserTypeUtil.potentiallyFreeze(getDataType(property)));
 				} else if (property.isClusterKeyColumn()) {
-					specification.clusteredKeyColumn(property.getColumnName(),
+					specification.clusteredKeyColumn(property.getRequiredColumnName(),
 							UserTypeUtil.potentiallyFreeze(getDataType(property)), property.getPrimaryKeyOrdering());
 				} else {
-					specification.column(property.getColumnName(),
-							UserTypeUtil.potentiallyFreeze(getDataType(property)));
+					specification.column(property.getRequiredColumnName(), UserTypeUtil.potentiallyFreeze(getDataType(property)));
 				}
 			}
 		}
@@ -466,8 +497,8 @@ public class CassandraMappingContext
 
 		for (CassandraPersistentProperty property : entity) {
 			// Use frozen literal to not resolve types from Cassandra; At this stage, they might be not created yet.
-			specification.field(property.getColumnName(),
-				getDataTypeWithUserTypeFactory(property, DataTypeProvider.FrozenLiteral));
+			specification.field(property.getRequiredColumnName(),
+					getDataTypeWithUserTypeFactory(property, DataTypeProvider.FrozenLiteral));
 		}
 
 		if (specification.getFields().isEmpty()) {
@@ -489,11 +520,18 @@ public class CassandraMappingContext
 	 */
 	public DataType getDataType(Class<?> type) {
 
-		return this.customConversions.getCustomWriteTarget(type)
-				.map(CassandraSimpleTypeHolder::getDataTypeFor)
+		return this.customConversions.getCustomWriteTarget(type).map(CassandraSimpleTypeHolder::getDataTypeFor)
 				.orElseGet(() -> getDataTypeFor(type));
 	}
 
+	public TupleType getTupleType(CassandraPersistentEntity<?> persistentEntity) {
+
+		Assert.notNull(persistentEntity, "CassandraPersistentEntity must not be null");
+		Assert.isTrue(persistentEntity.isTupleType(), "CassandraPersistentEntity is not a mapped tuple type");
+
+		return getTupleType(DataTypeProvider.EntityUserType, persistentEntity);
+	}
+
 	/**
 	 * Retrieve the data type of the property. Cassandra {@link DataType types} are determined using simple types and
 	 * configured {@link org.springframework.data.convert.CustomConversions}.
@@ -517,11 +555,10 @@ public class CassandraMappingContext
 
 			if (annotation.type() == Name.TUPLE) {
 
-				DataType[] dataTypes = Arrays.stream(annotation.typeArguments())
-					.map(CassandraSimpleTypeHolder::getDataTypeFor)
-					.toArray(DataType[]::new);
+				DataType[] dataTypes = Arrays.stream(annotation.typeArguments()).map(CassandraSimpleTypeHolder::getDataTypeFor)
+						.toArray(DataType[]::new);
 
-				return TupleType.of(ProtocolVersion.NEWEST_SUPPORTED, CodecRegistry.DEFAULT_INSTANCE, dataTypes);
+				return tupleTypeFactory.create(dataTypes);
 			}
 
 			if (annotation.type() == Name.UDT) {
@@ -547,46 +584,60 @@ public class CassandraMappingContext
 		return getDataTypeWithUserTypeFactory(property.getTypeInformation(), dataTypeProvider, property::getDataType);
 	}
 
-	private DataType getDataTypeWithUserTypeFactory(TypeInformation<?> typeInformation,
-			DataTypeProvider dataTypeProvider, Supplier<DataType> fallback) {
+	private DataType getDataTypeWithUserTypeFactory(TypeInformation<?> typeInformation, DataTypeProvider dataTypeProvider,
+			Supplier<DataType> fallback) {
 
-		BasicCassandraPersistentEntity<?> persistentEntity =
-				getPersistentEntity(typeInformation.getRequiredActualType());
+		BasicCassandraPersistentEntity<?> persistentEntity = getPersistentEntity(typeInformation.getRequiredActualType());
 
-		if (persistentEntity != null && persistentEntity.isUserDefinedType()) {
+		if (persistentEntity != null) {
 
-			DataType dataType = getUserDataType(typeInformation, dataTypeProvider.getDataType(persistentEntity));
+			if (persistentEntity.isUserDefinedType()) {
 
-			if (dataType != null) {
-				return dataType;
+				DataType dataType = getUserDataType(typeInformation, dataTypeProvider.getDataType(persistentEntity));
+
+				if (dataType != null) {
+					return dataType;
+				}
+			}
+
+			if (persistentEntity.isTupleType()) {
+				return getUserDataType(typeInformation, getTupleType(dataTypeProvider, persistentEntity));
 			}
 		}
 
-		Optional<DataType> customWriteTarget = this.customConversions
-				.getCustomWriteTarget(typeInformation.getType())
+		Optional<DataType> customWriteTarget = this.customConversions.getCustomWriteTarget(typeInformation.getType())
 				.map(CassandraSimpleTypeHolder::getDataTypeFor);
 
-		DataType dataType = customWriteTarget.orElseGet(() ->
-				this.customConversions.getCustomWriteTarget(typeInformation.getRequiredActualType().getType())
-					.filter(it -> !typeInformation.isMap())
-					.map(it -> {
+		DataType dataType = customWriteTarget
+				.orElseGet(() -> this.customConversions.getCustomWriteTarget(typeInformation.getRequiredActualType().getType())
+						.filter(it -> !typeInformation.isMap()).map(it -> {
 
-						if (typeInformation.isCollectionLike()) {
-							if (List.class.isAssignableFrom(typeInformation.getType())) {
-								return DataType.list(getDataTypeFor(it));
+							if (typeInformation.isCollectionLike()) {
+								if (List.class.isAssignableFrom(typeInformation.getType())) {
+									return DataType.list(getDataTypeFor(it));
+								}
+
+								if (Set.class.isAssignableFrom(typeInformation.getType())) {
+									return DataType.set(getDataTypeFor(it));
+								}
 							}
 
-							if (Set.class.isAssignableFrom(typeInformation.getType())) {
-								return DataType.set(getDataTypeFor(it));
-							}
-						}
+							return getDataTypeFor(it);
 
-						return getDataTypeFor(it);
-
-					}).orElse(null));
+						}).orElse(null));
 
 		return dataType != null ? dataType
-			: typeInformation.isMap() ? getMapDataType(typeInformation, dataTypeProvider) : fallback.get();
+				: typeInformation.isMap() ? getMapDataType(typeInformation, dataTypeProvider) : fallback.get();
+	}
+
+	private TupleType getTupleType(DataTypeProvider dataTypeProvider, CassandraPersistentEntity<?> persistentEntity) {
+
+		List<DataType> types = new ArrayList<>();
+		for (CassandraPersistentProperty persistentProperty : persistentEntity) {
+			types.add(getDataTypeWithUserTypeFactory(persistentProperty, dataTypeProvider));
+		}
+
+		return tupleTypeFactory.create(types);
 	}
 
 	@SuppressWarnings("all")
@@ -648,7 +699,7 @@ public class CassandraMappingContext
 
 			@Override
 			public DataType getDataType(CassandraPersistentEntity<?> entity) {
-				return entity.getUserType();
+				return entity.isTupleType() ? entity.getTupleType() : entity.getUserType();
 			}
 
 			@Override

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CassandraPersistentEntity.java

@@ -19,6 +19,7 @@ import org.springframework.data.cassandra.core.cql.CqlIdentifier;
 import org.springframework.data.mapping.PersistentEntity;
 import org.springframework.lang.Nullable;
 
+import com.datastax.driver.core.TupleType;
 import com.datastax.driver.core.UserType;
 
 /**
@@ -55,7 +56,7 @@ public interface CassandraPersistentEntity<T> extends PersistentEntity<T, Cassan
 	void setForceQuote(boolean forceQuote);
 
 	/**
-	 * @return {@literal true} if the type is a mapped user defined type
+	 * @return {@literal true} if the type is a mapped user defined type.
 	 * @since 1.5
 	 * @see UserDefinedType
 	 */
@@ -68,4 +69,19 @@ public interface CassandraPersistentEntity<T> extends PersistentEntity<T, Cassan
 	 */
 	@Nullable
 	UserType getUserType();
+
+	/**
+	 * @return {@literal true} if the type is a mapped tuple type.
+	 * @since 2.1
+	 * @see Tuple
+	 */
+	boolean isTupleType();
+
+	/**
+	 * @return the {@link TupleType} matching the data types from {@link BasicCassandraPersistentTupleProperty mapped
+	 *         tuple elements}.
+	 * @since 2.1
+	 */
+	@Nullable
+	TupleType getTupleType();
 }

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CassandraPersistentProperty.java

@@ -42,8 +42,49 @@ public interface CassandraPersistentProperty
 	/**
 	 * The name of the single column to which the property is persisted.
 	 */
+	@Nullable
 	CqlIdentifier getColumnName();
 
+	/**
+	 * The name of the single column to which the property is persisted.
+	 *
+	 * @throws IllegalStateException if the required column name is not available.
+	 * @since 2.1
+	 */
+	default CqlIdentifier getRequiredColumnName() {
+
+		CqlIdentifier columnName = getColumnName();
+
+		if (columnName == null) {
+			throw new IllegalStateException("No column name available for this persistent property");
+		}
+
+		return columnName;
+	}
+
+	/**
+	 * The name of the element ordinal to which the property is persisted when the owning type is a mapped tuple.
+	 */
+	@Nullable
+	Integer getOrdinal();
+
+	/**
+	 * The required element ordinal to which the property is persisted when the owning type is a mapped tuple.
+	 *
+	 * @throws IllegalStateException if the required ordinal is not available.
+	 * @since 2.1
+	 */
+	default int getRequiredOrdinal() {
+
+		Integer ordinal = getOrdinal();
+
+		if (ordinal == null) {
+			throw new IllegalStateException("No ordinal available for this persistent property");
+		}
+
+		return ordinal;
+	}
+
 	/**
 	 * The ordering (ascending or descending) for the column. Valid only for primary key columns; returns null for
 	 * non-primary key columns.

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CassandraPersistentPropertyComparator.java

@@ -92,6 +92,6 @@ public enum CassandraPersistentPropertyComparator implements Comparator<Cassandr
 		}
 
 		// else, neither property is a composite primary key nor a primary key; compare @Column annotations
-		return left.getColumnName().compareTo(right.getColumnName());
+		return left.getRequiredColumnName().compareTo(right.getRequiredColumnName());
 	}
 }

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CassandraPersistentTupleMetadataVerifier.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2018 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.cassandra.core.mapping;
+
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import org.springframework.data.mapping.MappingException;
+import org.springframework.util.StringUtils;
+
+/**
+ * Verifier for {@link CassandraPersistentEntity tuple entities}. Validates for a proper annotated domain classes to
+ * ensure the meta-model is suitable for {@link com.datastax.driver.core.TupleValue} mapping.
+ *
+ * @author Mark Paluch
+ * @since 2.1
+ */
+enum CassandraPersistentTupleMetadataVerifier implements CassandraPersistentEntityMetadataVerifier {
+
+	INSTANCE;
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.CassandraPersistentEntityMetadataVerifier#verify(org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity)
+	 */
+	public void verify(CassandraPersistentEntity<?> entity) throws MappingException {
+
+		if (entity.getType().isInterface() || !entity.isAnnotationPresent(Tuple.class)) {
+			return;
+		}
+
+		Set<Integer> ordinals = new TreeSet<>();
+
+		for (CassandraPersistentProperty tupleProperty : entity) {
+
+			if (tupleProperty.isTransient()) {
+				continue;
+			}
+
+			if (!ordinals.add(tupleProperty.getOrdinal())) {
+				throw new MappingException(
+						String.format("Duplicate ordinal [%d] in entity [%s]", tupleProperty.getOrdinal(), entity.getName()));
+			}
+		}
+
+		if (ordinals.isEmpty()) {
+			throw new MappingException(String.format(
+					"Mapped tuple contains no persistent elements annotated with @Element in entity [%s]", entity.getName()));
+		}
+
+		List<Integer> missingMappings = IntStream.range(0, ordinals.size()).boxed().collect(Collectors.toList());
+
+		missingMappings.removeAll(ordinals);
+
+		if (!missingMappings.isEmpty()) {
+			throw new MappingException(String.format("Mapped tuple has no ordinal mapping in entity [%s] for ordinal(s): %s",
+					entity.getName(), StringUtils.collectionToDelimitedString(missingMappings, ", ")));
+		}
+	}
+}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/CodecRegistryTupleTypeFactory.java

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2018 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.cassandra.core.mapping;
+
+import java.util.List;
+
+import org.springframework.util.Assert;
+
+import com.datastax.driver.core.CodecRegistry;
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.ProtocolVersion;
+import com.datastax.driver.core.TupleType;
+
+/**
+ * {@link CodecRegistry}-based {@link TupleTypeFactory} using
+ * {@link TupleType#of(ProtocolVersion, CodecRegistry, DataType...)} to create tuple types. {@link TupleType tuple
+ * types}.
+ *
+ * @author Mark Paluch
+ * @since 2.1
+ */
+public class CodecRegistryTupleTypeFactory implements TupleTypeFactory {
+
+	/**
+	 * Default {@link CodecRegistryTupleTypeFactory} using newest protocol versions and the default {@link CodecRegistry}.
+	 */
+	public static final CodecRegistryTupleTypeFactory DEFAULT = new CodecRegistryTupleTypeFactory();
+
+	private final ProtocolVersion protocolVersion;
+
+	private final CodecRegistry codecRegistry;
+
+	/**
+	 * Creates a new {@link CodecRegistryTupleTypeFactory} using newest protocol version and the default
+	 * {@link CodecRegistry}.
+	 */
+	private CodecRegistryTupleTypeFactory() {
+		this(ProtocolVersion.NEWEST_SUPPORTED, CodecRegistry.DEFAULT_INSTANCE);
+	}
+
+	/**
+	 * Creates a new {@link CodecRegistryTupleTypeFactory} given {@link ProtocolVersion} and {@link CodecRegistry}.
+	 *
+	 * @param protocolVersion must not be {@literal null}.
+	 * @param codecRegistry must not be {@literal null}.
+	 */
+	public CodecRegistryTupleTypeFactory(ProtocolVersion protocolVersion, CodecRegistry codecRegistry) {
+
+		Assert.notNull(protocolVersion, "ProtocolVersion must not be null");
+		Assert.notNull(codecRegistry, "CodecRegistry must not be null");
+
+		this.protocolVersion = protocolVersion;
+		this.codecRegistry = codecRegistry;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.TupleTypeFactory#create(java.util.List)
+	 */
+	@Override
+	public TupleType create(List<DataType> types) {
+		return create(types.toArray(new DataType[0]));
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.TupleTypeFactory#create(com.datastax.driver.core.DataType[])
+	 */
+	@Override
+	public TupleType create(DataType... types) {
+		return TupleType.of(protocolVersion, codecRegistry, types);
+	}
+}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/Element.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2018 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.cassandra.core.mapping;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Annotation to define an ordinal element index within a tuple. Ordinals map to fields or property accessors within a
+ * domain class. Element indexes within a tuple must be unique and consecutive beginning with the first index at
+ * {@literal 0}.
+ *
+ * @author Mark Paluch
+ * @since 2.1
+ * @see Tuple
+ */
+@Inherited
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ ElementType.FIELD, ElementType.METHOD })
+public @interface Element {
+
+	/**
+	 * @return ordinal index within a tuple. First index is {@literal 0}, must be greater or equal to zero.
+	 */
+	int value();
+}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/IndexSpecificationFactory.java

@@ -131,7 +131,7 @@ class IndexSpecificationFactory {
 			index = CreateIndexSpecification.createIndex();
 		}
 
-		return index.columnName(property.getColumnName());
+		return index.columnName(property.getRequiredColumnName());
 	}
 
 	private static CreateIndexSpecification createIndexSpecification(SASI annotation,
@@ -146,7 +146,7 @@ class IndexSpecificationFactory {
 		}
 
 		index.using("org.apache.cassandra.index.sasi.SASIIndex") //
-				.columnName(property.getColumnName()) //
+				.columnName(property.getRequiredColumnName()) //
 				.withOption("mode", annotation.indexMode().name());
 
 		long analyzerCount = INDEX_CONFIGURERS.keySet().stream().filter(property::isAnnotationPresent).count();

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/SimpleTupleTypeFactory.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2018 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.cassandra.core.mapping;
+
+import java.util.List;
+
+import org.springframework.util.Assert;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.TupleType;
+
+/**
+ * Default {@link TupleTypeFactory} implementation using Cluster {@link com.datastax.driver.core.Metadata} to create
+ * {@link TupleType tuple types}.
+ *
+ * @author Mark Paluch
+ * @since 2.1
+ */
+public class SimpleTupleTypeFactory implements TupleTypeFactory {
+
+	private final Cluster cluster;
+
+	/**
+	 * Creates a new {@link SimpleTupleTypeFactory} given {@link Cluster}.
+	 *
+	 * @param cluster must not be {@literal null}.
+	 */
+	public SimpleTupleTypeFactory(Cluster cluster) {
+
+		Assert.notNull(cluster, "Cluster must not be null");
+
+		this.cluster = cluster;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.cassandra.core.mapping.TupleTypeFactory#create(java.util.List)
+	 */
+	@Override
+	public TupleType create(List<DataType> types) {
+		return cluster.getMetadata().newTupleType(types);
+	}
+}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/Tuple.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2018 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.cassandra.core.mapping;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Identifies a domain object as Cassandra Tuple. Tuples use ordered fields to map their value to the actual property.
+ * <p/>
+ * A mapped tuple type is typically annotated with {@code @Tuple} and its properties/accessors are annotated with
+ * {@code @Element(0), @Element(1), ..., @Element(n)}.
+ * <p/>
+ * Example usage:
+ *
+ * <pre class="code">
+ * &#64;Tuple
+ * class Address {
+ *
+ * 	&#64;Element(0) String street;
+ *
+ * 	&#64;Element(1) &#64;CassandraType(type = Name.ASCII) String city;
+ *
+ * 	&#64;Element(2) int sortOrder;
+ * }
+ * </pre>
+ *
+ * @author Mark Paluch
+ * @since 2.1
+ * @see Element
+ */
+@Inherited
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ ElementType.TYPE })
+public @interface Tuple {
+}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/mapping/TupleTypeFactory.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2018 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.cassandra.core.mapping;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.springframework.util.Assert;
+
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.TupleType;
+
+/**
+ * Factory to create {@link TupleType} given {@link DataType tuple element types}. Primarily internal use.
+ *
+ * @author Mark Paluch
+ * @since 2.1
+ * @see SimpleTupleTypeFactory
+ * @see CodecRegistryTupleTypeFactory
+ */
+@FunctionalInterface
+public interface TupleTypeFactory {
+
+	/**
+	 * Create a {@link TupleType} representing the given {@link DataType tuple element types}.
+	 *
+	 * @param types must not be {@literal null} and not contain {@literal null} elements.
+	 * @return the {@link TupleType} representing the given {@link DataType tuple element types}.
+	 */
+	default TupleType create(DataType... types) {
+
+		Assert.notNull(types, "DataType must not be null");
+		Assert.noNullElements(types, "DataType must not contain null elements");
+
+		return create(Arrays.asList(types));
+	}
+
+	/**
+	 * Create a {@link TupleType} representing the given {@link DataType tuple element types}.
+	 *
+	 * @param types must not be {@literal null}.
+	 * @return the {@link TupleType} representing the given {@link DataType tuple element types}.
+	 */
+	TupleType create(List<DataType> types);
+}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverterMappedTupleUnitTests.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2018 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.cassandra.core.convert;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.springframework.data.cassandra.test.util.RowMockUtil.*;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+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.junit.MockitoJUnitRunner;
+import org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentEntity;
+import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
+import org.springframework.data.cassandra.core.mapping.Element;
+import org.springframework.data.cassandra.core.mapping.Tuple;
+import org.springframework.data.cassandra.test.util.RowMockUtil;
+
+import com.datastax.driver.core.Row;
+import com.datastax.driver.core.TupleValue;
+import com.datastax.driver.core.querybuilder.Insert;
+import com.datastax.driver.core.querybuilder.QueryBuilder;
+
+/**
+ * Unit tests for mapped tuples through {@link MappingCassandraConverter}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.Silent.class)
+public class MappingCassandraConverterMappedTupleUnitTests {
+
+	@Rule public final ExpectedException expectedException = ExpectedException.none();
+
+	Row rowMock;
+
+	CassandraMappingContext mappingContext;
+	MappingCassandraConverter mappingCassandraConverter;
+
+	@Before
+	public void setUp() {
+
+		mappingContext = new CassandraMappingContext();
+
+		mappingCassandraConverter = new MappingCassandraConverter(mappingContext);
+		mappingCassandraConverter.afterPropertiesSet();
+	}
+
+	@Test // DATACASS-523
+	public void shouldReadMappedTupleValue() {
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(MappedTuple.class);
+
+		TupleValue value = entity.getTupleType().newValue("hello", 1);
+
+		rowMock = RowMockUtil.newRowMock(column("tuple", value, entity.getTupleType()));
+
+		Person person = mappingCassandraConverter.read(Person.class, rowMock);
+
+		MappedTuple tuple = person.getTuple();
+
+		assertThat(tuple.getName()).isEqualTo("hello");
+		assertThat(tuple.getPosition()).isEqualTo(1);
+	}
+
+	@Test // DATACASS-523
+	public void shouldWriteMappedTuple() {
+
+		MappedTuple tuple = new MappedTuple("hello", 1);
+		Person person = new Person(tuple);
+
+		Insert insert = QueryBuilder.insertInto("table");
+
+		mappingCassandraConverter.write(person, insert);
+
+		assertThat(insert.toString()).contains("VALUES (('hello',1))");
+	}
+
+	@Data
+	@AllArgsConstructor
+	private static class Person {
+		MappedTuple tuple;
+	}
+
+	@Tuple
+	@Data
+	@AllArgsConstructor
+	private static class MappedTuple {
+
+		@Element(0) String name;
+		@Element(1) int position;
+	}
+
+}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/convert/MappingCassandraConverterTupleIntegrationTests.java

@@ -0,0 +1,200 @@
+/*
+ * Copyright 2018 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.cassandra.core.convert;
+
+import static org.assertj.core.api.Assertions.*;
+
+import lombok.Data;
+
+import java.util.Arrays;
+import java.util.Currency;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.cassandra.config.SchemaAction;
+import org.springframework.data.cassandra.core.cql.generator.CreateTableCqlGenerator;
+import org.springframework.data.cassandra.core.cql.generator.CreateUserTypeCqlGenerator;
+import org.springframework.data.cassandra.core.cql.keyspace.CreateTableSpecification;
+import org.springframework.data.cassandra.core.cql.keyspace.CreateUserTypeSpecification;
+import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
+import org.springframework.data.cassandra.core.mapping.Element;
+import org.springframework.data.cassandra.core.mapping.Table;
+import org.springframework.data.cassandra.core.mapping.Tuple;
+import org.springframework.data.cassandra.core.mapping.UserDefinedType;
+import org.springframework.data.cassandra.domain.AllPossibleTypes;
+import org.springframework.data.cassandra.repository.support.AbstractSpringDataEmbeddedCassandraIntegrationTest;
+import org.springframework.data.cassandra.repository.support.IntegrationTestConfig;
+import org.springframework.data.convert.CustomConversions;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.querybuilder.Insert;
+import com.datastax.driver.core.querybuilder.QueryBuilder;
+
+/**
+ * Integration tests for mapped tuple values through {@link MappingCassandraConverter}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration
+public class MappingCassandraConverterTupleIntegrationTests extends AbstractSpringDataEmbeddedCassandraIntegrationTest {
+
+	private static AtomicBoolean initialized = new AtomicBoolean();
+
+	@Configuration
+	public static class Config extends IntegrationTestConfig {
+
+		@Override
+		public SchemaAction getSchemaAction() {
+			return SchemaAction.NONE;
+		}
+
+		@Override
+		public String[] getEntityBasePackages() {
+			return new String[] { AllPossibleTypes.class.getPackage().getName() };
+		}
+
+		@Override
+		public CustomConversions customConversions() {
+			return new CassandraCustomConversions(
+					Arrays.asList(new StringToCurrencyConverter(), new CurrencyToStringConverter()));
+		}
+	}
+
+	@Autowired Session session;
+	@Autowired MappingCassandraConverter converter;
+
+	@Before
+	public void setUp() {
+
+		if (initialized.compareAndSet(false, true)) {
+
+			session.execute("DROP TYPE IF EXISTS address;");
+			session.execute("DROP TABLE IF EXISTS person;");
+
+			CassandraMappingContext mappingContext = converter.getMappingContext();
+
+			CreateUserTypeSpecification createAddress = mappingContext
+					.getCreateUserTypeSpecificationFor(mappingContext.getRequiredPersistentEntity(AddressUserType.class));
+
+			session.execute(CreateUserTypeCqlGenerator.toCql(createAddress));
+
+			CreateTableSpecification createPerson = mappingContext
+					.getCreateTableSpecificationFor(mappingContext.getRequiredPersistentEntity(Person.class));
+
+			session.execute(CreateTableCqlGenerator.toCql(createPerson));
+		} else {
+			session.execute("TRUNCATE person;");
+		}
+	}
+
+	@Test // DATACASS-523
+	public void shouldInsertRowWithComplexTuple() {
+
+		Person person = new Person();
+		person.setId("foo");
+
+		MappedTuple tuple = new MappedTuple();
+		AddressUserType userType = new AddressUserType();
+		userType.setZip("myzip");
+		tuple.setAddressUserType(userType);
+		tuple.setCurrency(Arrays.asList(Currency.getInstance("EUR"), Currency.getInstance("USD")));
+		tuple.setName("bar");
+
+		person.setMappedTuple(tuple);
+		person.setMappedTuples(Arrays.asList(tuple));
+
+		Insert insert = QueryBuilder.insertInto("person");
+		converter.write(person, insert);
+
+		session.execute(insert);
+	}
+
+	@Test // DATACASS-523
+	public void shouldReadRowWithComplexTuple() {
+
+		session.execute("INSERT INTO person (id,mappedtuple,mappedtuples) VALUES (" + //
+				"'foo'," //
+				+ "({zip:'myzip'},['EUR','USD'],'bar')," //
+				+ "[({zip:'myzip'},['EUR','USD'],'bar')]);\n");
+
+		ResultSet resultSet = session.execute("SELECT * FROM person;");
+
+		Person person = converter.read(Person.class, resultSet.one());
+
+		assertThat(person.getMappedTuples()).hasSize(1);
+		assertThat(person.getMappedTuple()).isNotNull();
+
+		MappedTuple mappedTuple = person.getMappedTuple();
+
+		assertThat(mappedTuple.getAddressUserType()).isNotNull();
+		assertThat(mappedTuple.getAddressUserType().getZip()).isEqualTo("myzip");
+		assertThat(mappedTuple.getName()).isEqualTo("bar");
+		assertThat(mappedTuple.getCurrency()).containsSequence(Currency.getInstance("EUR"), Currency.getInstance("USD"));
+	}
+
+	@Data
+	@Table
+	static class Person {
+
+		@Id private String id;
+
+		MappedTuple mappedTuple;
+		List<MappedTuple> mappedTuples;
+	}
+
+	@Data
+	@Tuple
+	static class MappedTuple {
+
+		@Element(0) AddressUserType addressUserType;
+		@Element(1) List<Currency> currency;
+		@Element(2) String name;
+
+	}
+
+	@UserDefinedType("address")
+	@Data
+	static class AddressUserType {
+		String zip;
+	}
+
+	private static class StringToCurrencyConverter implements Converter<String, Currency> {
+
+		@Override
+		public Currency convert(String source) {
+			return Currency.getInstance(source);
+		}
+	}
+
+	private static class CurrencyToStringConverter implements Converter<Currency, String> {
+
+		@Override
+		public String convert(Currency source) {
+			return source.getCurrencyCode();
+		}
+	}
+}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/convert/QueryMapperUnitTests.java

@@ -15,9 +15,11 @@
  */
 package org.springframework.data.cassandra.core.convert;
 
-import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.*;
 import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.*;
+
+import lombok.AllArgsConstructor;
 
 import java.util.Collection;
 import java.util.Collections;
@@ -26,19 +28,18 @@ import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 
-import lombok.AllArgsConstructor;
-
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
-
 import org.springframework.data.annotation.Id;
 import org.springframework.data.cassandra.core.cql.CqlIdentifier;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.Column;
+import org.springframework.data.cassandra.core.mapping.Element;
+import org.springframework.data.cassandra.core.mapping.Tuple;
 import org.springframework.data.cassandra.core.mapping.UserDefinedType;
 import org.springframework.data.cassandra.core.mapping.UserTypeResolver;
 import org.springframework.data.cassandra.core.query.ColumnName;
@@ -56,6 +57,7 @@ import org.springframework.data.domain.Sort.Direction;
 import org.springframework.data.domain.Sort.Order;
 
 import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.TupleValue;
 import com.datastax.driver.core.UDTValue;
 import com.datastax.driver.core.UserType;
 
@@ -324,6 +326,27 @@ public class QueryMapperUnitTests {
 		assertThat(mappedObject).contains("first_name");
 	}
 
+	@Test // DATACASS-523
+	public void shouldMapTuple() {
+
+		MappedTuple tuple = new MappedTuple("foo");
+
+		Filter filter = Filter.from(Criteria.where("tuple").is(tuple));
+
+		Filter mappedObject = queryMapper.getMappedObject(filter, mappingContext.getRequiredPersistentEntity(Person.class));
+
+		TupleValue tupleValue = mappingContext.getRequiredPersistentEntity(MappedTuple.class).getTupleType().newValue();
+		tupleValue.setString(0, "foo");
+		assertThat(mappedObject).contains(Criteria.where("tuple").is(tupleValue));
+	}
+
+	@Test(expected = IllegalArgumentException.class) // DATACASS-523
+	public void referencingTupleElementsInQueryShouldFail() {
+
+		queryMapper.getMappedObject(Filter.from(Criteria.where("tuple.zip").is("")),
+				mappingContext.getRequiredPersistentEntity(Person.class));
+	}
+
 	static class Person {
 
 		@Id String id;
@@ -336,9 +359,18 @@ public class QueryMapperUnitTests {
 
 		Integer number;
 
+		MappedTuple tuple;
+
 		@Column("first_name") String firstName;
 	}
 
+	@Tuple
+	@AllArgsConstructor
+	static class MappedTuple {
+
+		@Element(0) String zip;
+	}
+
 	@UserDefinedType
 	@AllArgsConstructor
 	static class Address {

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/convert/UpdateMapperUnitTests.java

@@ -15,8 +15,11 @@
  */
 package org.springframework.data.cassandra.core.convert;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.Mockito.when;
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
 
 import java.util.Collections;
 import java.util.Currency;
@@ -24,20 +27,18 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
-import lombok.AllArgsConstructor;
-import lombok.Data;
-
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
-
 import org.springframework.data.annotation.Id;
 import org.springframework.data.cassandra.core.cql.CqlIdentifier;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.Column;
+import org.springframework.data.cassandra.core.mapping.Element;
+import org.springframework.data.cassandra.core.mapping.Tuple;
 import org.springframework.data.cassandra.core.mapping.UserDefinedType;
 import org.springframework.data.cassandra.core.mapping.UserTypeResolver;
 import org.springframework.data.cassandra.core.query.Update;
@@ -222,6 +223,20 @@ public class UpdateMapperUnitTests {
 		assertThat(update.toString()).isEqualTo("number = number - 1");
 	}
 
+	@Test // DATACASS-523
+	public void shouldMapTuple() {
+
+		Update update = updateMapper.getMappedObject(Update.empty().set("tuple", new MappedTuple("foo")), persistentEntity);
+
+		assertThat(update.getUpdateOperations()).hasSize(1);
+		assertThat(update.toString()).isEqualTo("tuple = ('foo')");
+	}
+
+	@Test(expected = IllegalArgumentException.class) // DATACASS-523
+	public void referencingTupleElementsInQueryShouldFail() {
+		updateMapper.getMappedObject(Update.empty().set("tuple.zip", "bar"), persistentEntity);
+	}
+
 	static class Person {
 
 		@Id String id;
@@ -233,10 +248,17 @@ public class UpdateMapperUnitTests {
 		Currency currency;
 
 		Integer number;
+		MappedTuple tuple;
 
 		@Column("first_name") String firstName;
 	}
 
+	@Tuple
+	@AllArgsConstructor
+	static class MappedTuple {
+		@Element(0) String zip;
+	}
+
 	@Data
 	@UserDefinedType
 	@AllArgsConstructor

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/BasicCassandraPersistentPropertyUnitTests.java

@@ -42,7 +42,7 @@ public class BasicCassandraPersistentPropertyUnitTests {
 
 	@Test
 	public void usesAnnotatedColumnName() {
-		assertThat(getPropertyFor(Timeline.class, "text").getColumnName().toCql()).isEqualTo("message");
+		assertThat(getPropertyFor(Timeline.class, "text").getRequiredColumnName().toCql()).isEqualTo("message");
 	}
 
 	@Test
@@ -55,7 +55,7 @@ public class BasicCassandraPersistentPropertyUnitTests {
 
 	@Test
 	public void returnsPropertyNameForUnannotatedProperty() {
-		assertThat(getPropertyFor(Timeline.class, "time").getColumnName().toCql()).isEqualTo("time");
+		assertThat(getPropertyFor(Timeline.class, "time").getRequiredColumnName().toCql()).isEqualTo("time");
 	}
 
 	@Test // DATACASS-259
@@ -63,7 +63,7 @@ public class BasicCassandraPersistentPropertyUnitTests {
 
 		CassandraPersistentProperty persistentProperty = getPropertyFor(TypeWithComposedColumnAnnotation.class, "column");
 
-		assertThat(persistentProperty.getColumnName()).isEqualTo(CqlIdentifier.of("mycolumn", true));
+		assertThat(persistentProperty.getRequiredColumnName()).isEqualTo(CqlIdentifier.of("mycolumn", true));
 	}
 
 	@Test // DATACASS-259
@@ -72,7 +72,7 @@ public class BasicCassandraPersistentPropertyUnitTests {
 		CassandraPersistentProperty persistentProperty = getPropertyFor(TypeWithComposedPrimaryKeyAnnotation.class,
 				"column");
 
-		assertThat(persistentProperty.getColumnName()).isEqualTo(CqlIdentifier.of("primary-key", true));
+		assertThat(persistentProperty.getRequiredColumnName()).isEqualTo(CqlIdentifier.of("primary-key", true));
 		assertThat(persistentProperty.isIdProperty()).isTrue();
 	}
 
@@ -82,7 +82,7 @@ public class BasicCassandraPersistentPropertyUnitTests {
 		CassandraPersistentProperty persistentProperty = getPropertyFor(TypeWithComposedPrimaryKeyColumnAnnotation.class,
 				"column");
 
-		assertThat(persistentProperty.getColumnName()).isEqualTo(CqlIdentifier.of("mycolumn", true));
+		assertThat(persistentProperty.getRequiredColumnName()).isEqualTo(CqlIdentifier.of("mycolumn", true));
 		assertThat(persistentProperty.isPrimaryKeyColumn()).isTrue();
 	}
 
@@ -113,7 +113,6 @@ public class BasicCassandraPersistentPropertyUnitTests {
 		return new BasicCassandraPersistentProperty(Property.of(ClassTypeInformation.from(type), field), getEntity(type),
 				CassandraSimpleTypeHolder.HOLDER);
 	}
-
 	private <T> BasicCassandraPersistentEntity<T> getEntity(Class<T> type) {
 		return new BasicCassandraPersistentEntity<>(ClassTypeInformation.from(type));
 	}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/BasicCassandraPersistentTupleEntityUnitTests.java

@@ -0,0 +1,169 @@
+/*
+ * Copyright 2018 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.cassandra.core.mapping;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.springframework.data.annotation.Transient;
+import org.springframework.data.mapping.MappingException;
+
+import com.datastax.driver.core.CodecRegistry;
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.ProtocolVersion;
+import com.datastax.driver.core.TupleType;
+
+/**
+ * Unit tests for {@link BasicCassandraPersistentTupleEntity}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class BasicCassandraPersistentTupleEntityUnitTests {
+
+	CassandraMappingContext mappingContext = new CassandraMappingContext();
+
+	@Mock TupleTypeFactory tupleTypeFactory;
+
+	@Before
+	public void before() {
+		mappingContext.setTupleTypeFactory(tupleTypeFactory);
+	}
+
+	@Test // DATACASS-523
+	public void shouldCreatePersistentTupleEntity() {
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(Address.class);
+
+		entity.verify();
+	}
+
+	@Test // DATACASS-523
+	public void shouldCreateElementsInOrder() {
+
+		List<String> propertyNames = new ArrayList<>();
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(Address.class);
+		entity.verify();
+
+		entity.forEach(it -> {
+			propertyNames.add(it.getName());
+		});
+
+		assertThat(propertyNames).containsSequence("street", "city", "sortOrder");
+	}
+
+	@Test // DATACASS-523
+	public void shouldCreateTupleType() {
+
+		when(tupleTypeFactory.create(anyList())).thenReturn(TupleType.of(ProtocolVersion.NEWEST_SUPPORTED,
+				CodecRegistry.DEFAULT_INSTANCE, DataType.text(), DataType.text(), DataType.cint()));
+
+		BasicCassandraPersistentEntity<?> entity = mappingContext.getRequiredPersistentEntity(Address.class);
+		entity.verify();
+
+		entity.getTupleType();
+
+		verify(tupleTypeFactory).create(Arrays.asList(DataType.text(), DataType.text(), DataType.cint()));
+	}
+
+	@Test // DATACASS-523
+	public void shouldReportDuplicateMappings() {
+
+		assertThatThrownBy(() -> mappingContext.getRequiredPersistentEntity(DuplicateElement.class))
+				.isInstanceOf(MappingException.class).hasMessageContaining("Duplicate ordinal [0]");
+	}
+
+	@Test // DATACASS-523
+	public void shouldReportMissingOrdinalMappings() {
+
+		assertThatThrownBy(() -> mappingContext.getRequiredPersistentEntity(MissingElementOrdinals.class))
+				.isInstanceOf(MappingException.class).hasMessageContaining("Mapped tuple has no")
+				.hasMessageContaining("for ordinal(s): 0");
+	}
+
+	@Test // DATACASS-523
+	public void shouldReportNegativeOrdinalIndex() {
+
+		assertThatThrownBy(() -> mappingContext.getRequiredPersistentEntity(NegativeIndex.class))
+				.isInstanceOf(IllegalArgumentException.class)
+				.hasMessageContaining("Element ordinal must be greater or equal to zero for property [street] in entity");
+	}
+
+	@Test // DATACASS-523
+	public void shouldReportNoElements() {
+
+		assertThatThrownBy(() -> mappingContext.getRequiredPersistentEntity(NoElements.class))
+				.isInstanceOf(MappingException.class)
+				.hasMessageContaining("Mapped tuple contains no persistent elements annotated");
+	}
+
+	@Test // DATACASS-523
+	public void shouldReportMissingAnnotations() {
+
+		assertThatThrownBy(() -> mappingContext.getRequiredPersistentEntity(MissingAnnotation.class))
+				.isInstanceOf(MappingException.class)
+				.hasMessageContaining("Missing @Element annotation in mapped tuple type for property [street]");
+	}
+
+	@Tuple
+	static class Address {
+
+		@Element(1) String city;
+		@Element(0) String street;
+		@Element(2) int sortOrder;
+	}
+
+	@Tuple
+	static class DuplicateElement {
+
+		@Element(0) String street;
+		@Element(0) String city;
+	}
+
+	@Tuple
+	static class NegativeIndex {
+		@Element(-1) String street;
+	}
+
+	@Tuple
+	static class MissingElementOrdinals {
+
+		@Element(1) String street;
+		@Element(3) String city;
+	}
+
+	@Tuple
+	static class NoElements {
+		@Transient String springDataTransient;
+	}
+
+	@Tuple
+	static class MissingAnnotation {
+
+		String street;
+		String city;
+	}
+}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/BasicCassandraPersistentTuplePropertyUnitTests.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2018 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.cassandra.core.mapping;
+
+import static org.assertj.core.api.Assertions.*;
+
+import java.lang.reflect.Field;
+import java.util.Date;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.springframework.data.mapping.model.Property;
+import org.springframework.data.util.ClassTypeInformation;
+import org.springframework.util.ReflectionUtils;
+
+/**
+ * Unit tests for {@link BasicCassandraPersistentTupleEntity}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class BasicCassandraPersistentTuplePropertyUnitTests {
+
+	@Mock TupleTypeFactory tupleTypeFactory;
+
+	@Test // DATACASS-523
+	public void mappedTupleShouldNotReportColumnName() {
+
+		CassandraPersistentProperty property = getPropertyFor(MappedTuple.class, "date");
+
+		assertThat(property.getColumnName()).isNull();
+	}
+
+	@Test // DATACASS-523
+	public void mappedTupleShouldReportOrdinal() {
+
+		CassandraPersistentProperty property = getPropertyFor(MappedTuple.class, "time");
+
+		assertThat(property.getOrdinal()).isEqualTo(1);
+	}
+
+	private CassandraPersistentProperty getPropertyFor(Class<?> type, String fieldName) {
+
+		Field field = ReflectionUtils.findField(type, fieldName);
+
+		return new BasicCassandraPersistentTupleProperty(Property.of(ClassTypeInformation.from(type), field),
+				getEntity(type), CassandraSimpleTypeHolder.HOLDER);
+	}
+
+	private <T> BasicCassandraPersistentEntity<T> getEntity(Class<T> type) {
+		return new BasicCassandraPersistentTupleEntity<>(ClassTypeInformation.from(type), tupleTypeFactory);
+	}
+
+	@Tuple
+	static class MappedTuple {
+		@Element(0) Date date;
+		@Element(1) Date time;
+	}
+}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/CassandraMappingContextUnitTests.java

@@ -15,10 +15,8 @@
  */
 package org.springframework.data.cassandra.core.mapping;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.fail;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
 
 import java.io.Serializable;
 import java.util.Collection;
@@ -29,7 +27,6 @@ import java.util.NoSuchElementException;
 
 import org.junit.Before;
 import org.junit.Test;
-
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.annotation.Id;
@@ -252,14 +249,13 @@ public class CassandraMappingContextUnitTests {
 		UserType mappedudt = UserTypeBuilder.forName("mappedudt").withField("foo", DataType.ascii()).build();
 
 		this.mappingContext.setUserTypeResolver(typeName -> mappedudt);
-		this.mappingContext.setCustomConversions(new CassandraCustomConversions(
-				Collections.singletonList(HumanToStringConverter.INSTANCE)));
+		this.mappingContext.setCustomConversions(
+				new CassandraCustomConversions(Collections.singletonList(HumanToStringConverter.INSTANCE)));
 
-		CassandraPersistentEntity<?> persistentEntity =
-				this.mappingContext.getRequiredPersistentEntity(WithMapOfMixedTypes.class);
+		CassandraPersistentEntity<?> persistentEntity = this.mappingContext
+				.getRequiredPersistentEntity(WithMapOfMixedTypes.class);
 
-		CreateTableSpecification tableSpecification =
-				this.mappingContext.getCreateTableSpecificationFor(persistentEntity);
+		CreateTableSpecification tableSpecification = this.mappingContext.getCreateTableSpecificationFor(persistentEntity);
 
 		assertThat(tableSpecification.getColumns()).hasSize(2);
 
@@ -369,16 +365,15 @@ public class CassandraMappingContextUnitTests {
 
 	@Test(expected = InvalidDataAccessApiUsageException.class) // DATACASS-284
 	public void shouldRejectUntypedTuples() {
-		this.mappingContext.getCreateTableSpecificationFor(
-				this.mappingContext.getRequiredPersistentEntity(UntypedTupleEntity.class));
+		this.mappingContext
+				.getCreateTableSpecificationFor(this.mappingContext.getRequiredPersistentEntity(UntypedTupleEntity.class));
 	}
 
 	@Test // DATACASS-284
 	public void shouldCreateTableForTypedTupleType() {
 
-		CreateTableSpecification tableSpecification =
-				this.mappingContext.getCreateTableSpecificationFor(
-					this.mappingContext.getRequiredPersistentEntity(TypedTupleEntity.class));
+		CreateTableSpecification tableSpecification = this.mappingContext
+				.getCreateTableSpecificationFor(this.mappingContext.getRequiredPersistentEntity(TypedTupleEntity.class));
 
 		assertThat(tableSpecification.getColumns()).hasSize(2);
 
@@ -477,6 +472,18 @@ public class CassandraMappingContextUnitTests {
 		assertThat(mappingContext.getTableEntities()).hasSize(0);
 	}
 
+	@Test // DATACASS-523
+	public void shouldCreateMappedTupleType() {
+
+		CassandraPersistentEntity<?> persistentEntity = mappingContext.getRequiredPersistentEntity(MappedTuple.class);
+
+		assertThat(persistentEntity).isInstanceOf(BasicCassandraPersistentTupleEntity.class);
+
+		assertThat(mappingContext.getUserDefinedTypeEntities()).isEmpty();
+		assertThat(mappingContext.getPersistentEntities()).hasSize(1);
+		assertThat(mappingContext.getTableEntities()).isEmpty();
+	}
+
 	@Test // DATACASS-172
 	public void getNonPrimaryKeyEntitiesShouldNotContainUdt() {
 
@@ -624,6 +631,11 @@ public class CassandraMappingContextUnitTests {
 		@PrimaryKey String key;
 	}
 
+	@Tuple
+	private static class MappedTuple {
+		@Element(0) String name;
+	}
+
 	@UserDefinedType
 	private static class MappedUdt {}
 

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/CassandraPersistentPropertyComparatorUnitTests.java

@@ -160,17 +160,17 @@ public class CassandraPersistentPropertyComparatorUnitTests {
 		when(left.isPrimaryKeyColumn()).thenReturn(true);
 		when(right.isCompositePrimaryKey()).thenReturn(true);
 		when(right.isPrimaryKeyColumn()).thenReturn(false);
-		when(left.getColumnName()).thenReturn(CqlIdentifier.of("left"));
-		when(right.getColumnName()).thenReturn(CqlIdentifier.of("right"));
+		when(left.getRequiredColumnName()).thenReturn(CqlIdentifier.of("left"));
+		when(right.getRequiredColumnName()).thenReturn(CqlIdentifier.of("right"));
 
 		assertThat(INSTANCE.compare(left, right)).isLessThan(0);
 
 		verify(left, times(1)).isCompositePrimaryKey();
 		verify(left, times(1)).isPrimaryKeyColumn();
-		verify(left, times(1)).getColumnName();
+		verify(left, times(1)).getRequiredColumnName();
 		verify(right, times(1)).isCompositePrimaryKey();
 		verify(right, times(1)).isPrimaryKeyColumn();
-		verify(right, times(1)).getColumnName();
+		verify(right, times(1)).getRequiredColumnName();
 	}
 
 	@Test // DATACASS-352

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/CreateTableSpecificationBasicCassandraMappingContextUnitTests.java

@@ -40,9 +40,12 @@ import org.springframework.data.cassandra.core.cql.keyspace.CreateTableSpecifica
 import org.springframework.data.cassandra.domain.AllPossibleTypes;
 import org.springframework.util.StringUtils;
 
+import com.datastax.driver.core.CodecRegistry;
 import com.datastax.driver.core.DataType;
 import com.datastax.driver.core.DataType.CollectionType;
 import com.datastax.driver.core.DataType.Name;
+import com.datastax.driver.core.ProtocolVersion;
+import com.datastax.driver.core.TupleType;
 import com.datastax.driver.core.UDTValue;
 import com.datastax.driver.core.UserType;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -66,6 +69,8 @@ public class CreateTableSpecificationBasicCassandraMappingContextUnitTests {
 
 		CassandraCustomConversions customConversions = new CassandraCustomConversions(converters);
 		ctx.setCustomConversions(customConversions);
+		ctx.setTupleTypeFactory(types -> TupleType.of(ProtocolVersion.NEWEST_SUPPORTED, CodecRegistry.DEFAULT_INSTANCE,
+				types.toArray(new DataType[0])));
 	}
 
 	@Test // DATACASS-296
@@ -275,7 +280,7 @@ public class CreateTableSpecificationBasicCassandraMappingContextUnitTests {
 	}
 
 	@Test // DATACASS-172
-	public void columnsShouldMapToMapped() {
+	public void columnsShouldMapToMappedUserType() {
 
 		final UserType mappedUdt = mock(UserType.class, "mappedudt");
 
@@ -296,6 +301,36 @@ public class CreateTableSpecificationBasicCassandraMappingContextUnitTests {
 		assertThat(getColumnType("udtToString", specification)).isEqualTo(DataType.map(mappedUdt, DataType.varchar()));
 	}
 
+	@Test // DATACASS-523
+	public void columnsShouldMapToTuple() {
+
+		UserType mappedUdt = mock(UserType.class, "mappedudt");
+		UserType human_udt = mock(UserType.class, "human_udt");
+
+		when(mappedUdt.asFunctionParameterString()).thenReturn("mappedudt");
+		when(human_udt.asFunctionParameterString()).thenReturn("human_udt");
+
+		ctx.setUserTypeResolver(typeName -> {
+
+			if (typeName.toCql().equals(mappedUdt.toString())) {
+				return mappedUdt;
+			}
+
+			if (typeName.toCql().equals(human_udt.toString())) {
+				return human_udt;
+			}
+			return null;
+		});
+
+		CreateTableSpecification specification = getCreateTableSpecificationFor(WithMappedTuple.class);
+
+		assertThat(getColumnType("mappedTuple", specification).toString())
+				.isEqualTo("frozen<tuple<mappedudt, human_udt, text>>");
+
+		assertThat(getColumnType("mappedTuples", specification).toString())
+				.isEqualTo("list<frozen<tuple<mappedudt, human_udt, text>>>");
+	}
+
 	private CreateTableSpecification getCreateTableSpecificationFor(Class<?> persistentEntityClass) {
 
 		CassandraCustomConversions customConversions = new CassandraCustomConversions(Collections.emptyList());
@@ -335,6 +370,23 @@ public class CreateTableSpecificationBasicCassandraMappingContextUnitTests {
 		@CassandraType(type = Name.SET, typeArguments = Name.BIGINT) List<Human> enemies;
 	}
 
+	@Data
+	@Table
+	private static class WithMappedTuple {
+
+		@Id String id;
+		MappedTuple mappedTuple;
+		List<MappedTuple> mappedTuples;
+	}
+
+	@Tuple
+	private static class MappedTuple {
+
+		@Element(0) MappedUdt mappedUdt;
+		@Element(1) @CassandraType(type = Name.UDT, userTypeName = "human_udt") UDTValue human;
+		@Element(2) String text;
+	}
+
 	@Data
 	@Table
 	private static class WithUdtFields {

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/mapping/SimpleTupleTypeFactoryUnitTests.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2018 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.cassandra.core.mapping;
+
+import static org.mockito.Mockito.*;
+
+import java.util.Collections;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.Metadata;
+
+/**
+ * Unit tests for {@link SimpleTupleTypeFactory}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class SimpleTupleTypeFactoryUnitTests {
+
+	@Mock Cluster cluster;
+	@Mock Metadata metadata;
+
+	@Test // DATACASS-523
+	public void shouldCreateTupleTypes() {
+
+		when(cluster.getMetadata()).thenReturn(metadata);
+
+		new SimpleTupleTypeFactory(cluster).create(DataType.varchar());
+
+		verify(metadata).newTupleType(Collections.singletonList(DataType.varchar()));
+	}
+}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/test/util/RowMockUtil.java

@@ -77,6 +77,8 @@ public class RowMockUtil {
 		when(rowMock.getTimestamp(anyInt()))
 				.thenAnswer(invocation -> columns[(Integer) invocation.getArguments()[0]].value);
 		when(rowMock.getUUID(anyInt())).thenAnswer(invocation -> columns[(Integer) invocation.getArguments()[0]].value);
+		when(rowMock.getTupleValue(anyInt()))
+				.thenAnswer(invocation -> columns[(Integer) invocation.getArguments()[0]].value);
 
 		return rowMock;
 	}