Add support for direct and nested projections.

Projections are now fully supported from within the CassandraConverter that can materialize DTO and interface projections without using intermediate entities.

Closes: #1202

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

@@ -55,6 +55,7 @@ import org.springframework.data.cassandra.core.mapping.event.CassandraMappingEve
 import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.data.domain.Slice;
 import org.springframework.data.mapping.callback.EntityCallbacks;
+import org.springframework.data.projection.EntityProjection;
 import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 import org.springframework.data.util.Streamable;
@@ -119,8 +120,6 @@ public class AsyncCassandraTemplate
 
 	private final EntityOperations entityOperations;
 
-	private final SpelAwareProxyProjectionFactory projectionFactory;
-
 	private final StatementFactory statementFactory;
 
 	private @Nullable ApplicationEventPublisher eventPublisher;
@@ -186,9 +185,8 @@ public class AsyncCassandraTemplate
 
 		this.converter = converter;
 		this.cqlOperations = asyncCqlTemplate;
-		this.entityOperations = new EntityOperations(converter.getMappingContext());
+		this.entityOperations = new EntityOperations(converter);
 		this.exceptionTranslator = asyncCqlTemplate.getExceptionTranslator();
-		this.projectionFactory = new SpelAwareProxyProjectionFactory();
 		this.statementFactory = new StatementFactory(converter);
 	}
 
@@ -209,9 +207,6 @@ public class AsyncCassandraTemplate
 		if (entityCallbacks == null) {
 			setEntityCallbacks(EntityCallbacks.create(applicationContext));
 		}
-
-		projectionFactory.setBeanFactory(applicationContext);
-		projectionFactory.setBeanClassLoader(applicationContext.getClassLoader());
 	}
 
 	/**
@@ -284,9 +279,11 @@ public class AsyncCassandraTemplate
 	 *         projections.
 	 * @see org.springframework.data.projection.SpelAwareProxyProjectionFactory
 	 * @since 2.1
+	 * @deprecated since 3.4, use {@link CassandraConverter#getProjectionFactory()} instead.
 	 */
+	@Deprecated
 	protected SpelAwareProxyProjectionFactory getProjectionFactory() {
-		return this.projectionFactory;
+		return (SpelAwareProxyProjectionFactory) getConverter().getProjectionFactory();
 	}
 
 	private CassandraPersistentEntity<?> getRequiredPersistentEntity(Class<?> entityType) {
@@ -953,15 +950,13 @@ public class AsyncCassandraTemplate
 	@SuppressWarnings("unchecked")
 	private <T> Function<Row, T> getMapper(Class<?> entityType, Class<T> targetType, CqlIdentifier tableName) {
 
-		Class<?> typeToRead = resolveTypeToRead(entityType, targetType);
+		EntityProjection<T, ?> projection = entityOperations.introspectProjection(targetType, entityType);
 
 		return row -> {
 
 			maybeEmitEvent(new AfterLoadEvent<>(row, targetType, tableName));
 
-			Object source = getConverter().read(typeToRead, row);
-
-			T result = (T) (targetType.isInterface() ? getProjectionFactory().createProjection(targetType, source) : source);
+			T result = getConverter().project(projection, row);
 
 			if (result != null) {
 				maybeEmitEvent(new AfterConvertEvent<>(row, result, tableName));
@@ -971,10 +966,6 @@ public class AsyncCassandraTemplate
 		};
 	}
 
-	private Class<?> resolveTypeToRead(Class<?> entityType, Class<?> targetType) {
-		return targetType.isInterface() || targetType.isAssignableFrom(entityType) ? entityType : targetType;
-	}
-
 	private static MappingCassandraConverter newConverter(CqlSession session) {
 
 		MappingCassandraConverter converter = new MappingCassandraConverter();

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

@@ -55,6 +55,7 @@ import org.springframework.data.cassandra.core.query.Columns;
 import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.data.domain.Slice;
 import org.springframework.data.mapping.callback.EntityCallbacks;
+import org.springframework.data.projection.EntityProjection;
 import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 import org.springframework.lang.Nullable;
@@ -115,8 +116,6 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 
 	private final EntityOperations entityOperations;
 
-	private final SpelAwareProxyProjectionFactory projectionFactory;
-
 	private final StatementFactory statementFactory;
 
 	private @Nullable ApplicationEventPublisher eventPublisher;
@@ -182,8 +181,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 
 		this.converter = converter;
 		this.cqlOperations = cqlOperations;
-		this.entityOperations = new EntityOperations(converter.getMappingContext());
-		this.projectionFactory = new SpelAwareProxyProjectionFactory();
+		this.entityOperations = new EntityOperations(converter);
 		this.statementFactory = new StatementFactory(new QueryMapper(converter), new UpdateMapper(converter));
 	}
 
@@ -212,9 +210,6 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		if (entityCallbacks == null) {
 			setEntityCallbacks(EntityCallbacks.create(applicationContext));
 		}
-
-		projectionFactory.setBeanFactory(applicationContext);
-		projectionFactory.setBeanClassLoader(applicationContext.getClassLoader());
 	}
 
 	/**
@@ -287,9 +282,10 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 	 *         projections.
 	 * @see org.springframework.data.projection.SpelAwareProxyProjectionFactory
 	 * @since 2.1
+	 * @deprecated since 3.4, use {@link CassandraConverter#getProjectionFactory()} instead.
 	 */
 	protected SpelAwareProxyProjectionFactory getProjectionFactory() {
-		return this.projectionFactory;
+		return (SpelAwareProxyProjectionFactory) getConverter().getProjectionFactory();
 	}
 
 	private CassandraPersistentEntity<?> getRequiredPersistentEntity(Class<?> entityType) {
@@ -378,7 +374,8 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		Assert.notNull(statement, "Statement must not be null");
 		Assert.notNull(entityClass, "Entity type must not be null");
 
-		Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
+		Function<Row, T> mapper = getMapper(EntityProjection.nonProjecting(entityClass),
+				EntityQueryUtils.getTableName(statement));
 
 		return doQuery(statement, (row, rowNum) -> mapper.apply(row));
 	}
@@ -404,7 +401,8 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 
 		ResultSet resultSet = doQueryForResultSet(statement);
 
-		Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
+		Function<Row, T> mapper = getMapper(EntityProjection.nonProjecting(entityClass),
+				EntityQueryUtils.getTableName(statement));
 
 		return EntityQueryUtils.readSlice(resultSet, (row, rowNum) -> mapper.apply(row), 0,
 				getEffectivePageSize(statement));
@@ -419,7 +417,8 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		Assert.notNull(statement, "Statement must not be null");
 		Assert.notNull(entityClass, "Entity type must not be null");
 
-		Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
+		Function<Row, T> mapper = getMapper(EntityProjection.nonProjecting(entityClass),
+				EntityQueryUtils.getTableName(statement));
 		return doQueryForStream(statement, (row, rowNum) -> mapper.apply(row));
 	}
 
@@ -442,14 +441,14 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 	<T> List<T> doSelect(Query query, Class<?> entityClass, CqlIdentifier tableName, Class<T> returnType) {
 
 		CassandraPersistentEntity<?> entity = getRequiredPersistentEntity(entityClass);
-
-		Columns columns = getStatementFactory().computeColumnsForProjection(query.getColumns(), entity, returnType);
+		EntityProjection<T, ?> projection = entityOperations.introspectProjection(returnType, entityClass);
+		Columns columns = getStatementFactory().computeColumnsForProjection(projection, query.getColumns(), entity,
+				returnType);
 
 		Query queryToUse = query.columns(columns);
 
 		StatementBuilder<Select> select = getStatementFactory().select(queryToUse, entity, tableName);
-
-		Function<Row, T> mapper = getMapper(entityClass, returnType, tableName);
+		Function<Row, T> mapper = getMapper(projection, tableName);
 
 		return doQuery(select.build(), (row, rowNum) -> mapper.apply(row));
 	}
@@ -495,8 +494,9 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 
 		StatementBuilder<Select> select = getStatementFactory().select(query, getRequiredPersistentEntity(entityClass),
 				tableName);
+		EntityProjection<T, ?> projection = entityOperations.introspectProjection(returnType, entityClass);
 
-		Function<Row, T> mapper = getMapper(entityClass, returnType, tableName);
+		Function<Row, T> mapper = getMapper(projection, tableName);
 		return doQueryForStream(select.build(), (row, rowNum) -> mapper.apply(row));
 	}
 
@@ -639,7 +639,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		CassandraPersistentEntity<?> entity = getRequiredPersistentEntity(entityClass);
 		CqlIdentifier tableName = entity.getTableName();
 		StatementBuilder<Select> select = getStatementFactory().selectOneById(id, entity, tableName);
-		Function<Row, T> mapper = getMapper(entityClass, entityClass, tableName);
+		Function<Row, T> mapper = getMapper(EntityProjection.nonProjecting(entityClass), tableName);
 		List<T> result = doQuery(select.build(), (row, rowNum) -> mapper.apply(row));
 
 		return result.isEmpty() ? null : result.get(0);
@@ -999,17 +999,15 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 	}
 
 	@SuppressWarnings("unchecked")
-	private <T> Function<Row, T> getMapper(Class<?> entityType, Class<T> targetType, CqlIdentifier tableName) {
+	private <T> Function<Row, T> getMapper(EntityProjection<T, ?> projection, CqlIdentifier tableName) {
 
-		Class<?> typeToRead = resolveTypeToRead(entityType, targetType);
+		Class<T> targetType = projection.getMappedType().getType();
 
 		return row -> {
 
 			maybeEmitEvent(new AfterLoadEvent<>(row, targetType, tableName));
 
-			Object source = getConverter().read(typeToRead, row);
-
-			T result = (T) (targetType.isInterface() ? getProjectionFactory().createProjection(targetType, source) : source);
+			T result = getConverter().project(projection, row);
 
 			if (result != null) {
 				maybeEmitEvent(new AfterConvertEvent<>(row, result, tableName));
@@ -1019,10 +1017,6 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		};
 	}
 
-	private Class<?> resolveTypeToRead(Class<?> entityType, Class<?> targetType) {
-		return targetType.isInterface() || targetType.isAssignableFrom(entityType) ? entityType : targetType;
-	}
-
 	private static MappingCassandraConverter newConverter(CqlSession session) {
 
 		MappingCassandraConverter converter = new MappingCassandraConverter();

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

@@ -16,12 +16,17 @@
 package org.springframework.data.cassandra.core;
 
 import org.springframework.core.convert.ConversionService;
+import org.springframework.data.cassandra.core.convert.CassandraConverter;
 import org.springframework.data.cassandra.core.cql.util.StatementBuilder;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
+import org.springframework.data.convert.CustomConversions;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
+import org.springframework.data.projection.EntityProjection;
+import org.springframework.data.projection.EntityProjectionIntrospector;
+import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
@@ -44,10 +49,19 @@ import com.datastax.oss.driver.api.querybuilder.update.Update;
 class EntityOperations {
 
 	private final MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext;
+	private final EntityProjectionIntrospector introspector;
 
-	public EntityOperations(
-			MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> mappingContext) {
-		this.mappingContext = mappingContext;
+	EntityOperations(CassandraConverter converter) {
+		this(converter.getMappingContext(), converter.getCustomConversions(), converter.getProjectionFactory());
+	}
+
+	EntityOperations(MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> context,
+			CustomConversions conversions, ProjectionFactory projectionFactory) {
+		this.mappingContext = context;
+		this.introspector = EntityProjectionIntrospector.create(projectionFactory,
+				EntityProjectionIntrospector.ProjectionPredicate.typeHierarchy()
+						.and(((target, underlyingType) -> !conversions.isSimpleType(target))),
+				context);
 	}
 
 	/**
@@ -99,6 +113,9 @@ class EntityOperations {
 		return getRequiredPersistentEntity(entityClass).getTableName();
 	}
 
+	public <M, D> EntityProjection<M, D> introspectProjection(Class<M> resultType, Class<D> entityType) {
+		return introspector.introspect(resultType, entityType);
+	}
 
 	protected MappingContext<? extends CassandraPersistentEntity<?>, CassandraPersistentProperty> getMappingContext() {
 		return this.mappingContext;

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

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.cassandra.core;
 
+import org.springframework.data.projection.EntityProjection;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.SynchronousSink;
@@ -121,8 +122,6 @@ public class ReactiveCassandraTemplate
 
 	private final EntityOperations entityOperations;
 
-	private final SpelAwareProxyProjectionFactory projectionFactory;
-
 	private final StatementFactory statementFactory;
 
 	private @Nullable ApplicationEventPublisher eventPublisher;
@@ -189,8 +188,7 @@ public class ReactiveCassandraTemplate
 
 		this.converter = converter;
 		this.cqlOperations = reactiveCqlOperations;
-		this.entityOperations = new EntityOperations(converter.getMappingContext());
-		this.projectionFactory = new SpelAwareProxyProjectionFactory();
+		this.entityOperations = new EntityOperations(converter);
 		this.statementFactory = new StatementFactory(converter);
 	}
 
@@ -219,9 +217,6 @@ public class ReactiveCassandraTemplate
 		if (entityCallbacks == null) {
 			setEntityCallbacks(ReactiveEntityCallbacks.create(applicationContext));
 		}
-
-		projectionFactory.setBeanFactory(applicationContext);
-		projectionFactory.setBeanClassLoader(applicationContext.getClassLoader());
 	}
 
 	/**
@@ -294,9 +289,11 @@ public class ReactiveCassandraTemplate
 	 *         projections.
 	 * @see org.springframework.data.projection.SpelAwareProxyProjectionFactory
 	 * @since 2.1
+	 * @deprecated since 3.4, use {@link CassandraConverter#getProjectionFactory()} instead.
 	 */
+	@Deprecated
 	protected SpelAwareProxyProjectionFactory getProjectionFactory() {
-		return this.projectionFactory;
+		return (SpelAwareProxyProjectionFactory) getConverter().getProjectionFactory();
 	}
 
 	private CassandraPersistentEntity<?> getRequiredPersistentEntity(Class<?> entityType) {
@@ -365,7 +362,8 @@ public class ReactiveCassandraTemplate
 		Assert.notNull(statement, "Statement must not be null");
 		Assert.notNull(entityClass, "Entity type must not be null");
 
-		Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
+		Function<Row, T> mapper = getMapper(EntityProjection.nonProjecting(entityClass),
+				EntityQueryUtils.getTableName(statement));
 
 		return doQuery(statement, (row, rowNum) -> mapper.apply(row));
 	}
@@ -421,15 +419,15 @@ public class ReactiveCassandraTemplate
 	<T> Flux<T> doSelect(Query query, Class<?> entityClass, CqlIdentifier tableName, Class<T> returnType) {
 
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entityClass);
-
-		Columns columns = getStatementFactory().computeColumnsForProjection(query.getColumns(), persistentEntity,
+		EntityProjection<T, ?> projection = entityOperations.introspectProjection(returnType, entityClass);
+		Columns columns = getStatementFactory().computeColumnsForProjection(projection, query.getColumns(),
+				persistentEntity,
 				returnType);
 
 		Query queryToUse = query.columns(columns);
 
 		StatementBuilder<Select> select = getStatementFactory().select(queryToUse, persistentEntity, tableName);
-
-		Function<Row, T> mapper = getMapper(entityClass, returnType, tableName);
+		Function<Row, T> mapper = getMapper(projection, tableName);
 
 		return doQuery(select.build(), (row, rowNum) -> mapper.apply(row));
 	}
@@ -956,17 +954,15 @@ public class ReactiveCassandraTemplate
 	}
 
 	@SuppressWarnings("unchecked")
-	private <T> Function<Row, T> getMapper(Class<?> entityType, Class<T> targetType, CqlIdentifier tableName) {
+	private <T> Function<Row, T> getMapper(EntityProjection<T, ?> projection, CqlIdentifier tableName) {
 
-		Class<?> typeToRead = resolveTypeToRead(entityType, targetType);
+		Class<T> targetType = projection.getMappedType().getType();
 
 		return row -> {
 
 			maybeEmitEvent(new AfterLoadEvent<>(row, targetType, tableName));
 
-			Object source = getConverter().read(typeToRead, row);
-
-			T result = (T) (targetType.isInterface() ? getProjectionFactory().createProjection(targetType, source) : source);
+			T result = getConverter().project(projection, row);
 
 			if (result != null) {
 				maybeEmitEvent(new AfterConvertEvent<>(row, result, tableName));
@@ -981,10 +977,6 @@ public class ReactiveCassandraTemplate
 				.map(rows -> new WriteResult(resultSet.getAllExecutionInfo(), resultSet.wasApplied(), rows));
 	}
 
-	private Class<?> resolveTypeToRead(Class<?> entityType, Class<?> targetType) {
-		return targetType.isInterface() || targetType.isAssignableFrom(entityType) ? entityType : targetType;
-	}
-
 	private static MappingCassandraConverter newConverter(ReactiveSession session) {
 
 		MappingCassandraConverter converter = new MappingCassandraConverter();

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

@@ -37,6 +37,8 @@ import org.springframework.data.cassandra.core.cql.WriteOptions;
 import org.springframework.data.cassandra.core.cql.util.StatementBuilder;
 import org.springframework.data.cassandra.core.cql.util.TermFactory;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
+import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
+import org.springframework.data.cassandra.core.mapping.PersistentPropertyTranslator;
 import org.springframework.data.cassandra.core.query.Columns;
 import org.springframework.data.cassandra.core.query.Columns.ColumnSelector;
 import org.springframework.data.cassandra.core.query.Columns.FunctionCall;
@@ -57,11 +59,10 @@ import org.springframework.data.cassandra.core.query.Update.SetAtKeyOp;
 import org.springframework.data.cassandra.core.query.Update.SetOp;
 import org.springframework.data.convert.EntityWriter;
 import org.springframework.data.domain.Sort;
-import org.springframework.data.mapping.PersistentEntity;
 import org.springframework.data.mapping.PersistentProperty;
-import org.springframework.data.projection.ProjectionFactory;
+import org.springframework.data.projection.EntityProjection;
 import org.springframework.data.projection.ProjectionInformation;
-import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
+import org.springframework.data.util.Predicates;
 import org.springframework.lang.NonNull;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -106,8 +107,6 @@ public class StatementFactory {
 
 	private final UpdateMapper updateMapper;
 
-	private final ProjectionFactory projectionFactory = new SpelAwareProxyProjectionFactory();
-
 	/**
 	 * Create {@link StatementFactory} given {@link CassandraConverter}.
 	 *
@@ -560,22 +559,39 @@ public class StatementFactory {
 	 * {@literal closed interface projection}.
 	 *
 	 * @param columns must not be {@literal null}.
-	 * @param persistentEntity must not be {@literal null}.
+	 * @param domainType must not be {@literal null}.
 	 * @param returnType must not be {@literal null}.
 	 * @return {@link Columns} with columns to be included.
 	 * @since 2.2
 	 */
-	Columns computeColumnsForProjection(Columns columns, PersistentEntity<?, ?> persistentEntity, Class<?> returnType) {
+	Columns computeColumnsForProjection(EntityProjection<?, ?> projection, Columns columns,
+			CassandraPersistentEntity<?> domainType, Class<?> returnType) {
 
-		if (!columns.isEmpty() || ClassUtils.isAssignable(persistentEntity.getType(), returnType)) {
+		if (!columns.isEmpty() || ClassUtils.isAssignable(domainType.getType(), returnType)) {
 			return columns;
 		}
 
+		if (projection.getMappedType().getType().isInterface()) {
+			projection.forEach(propertyPath -> columns.include(propertyPath.getPropertyPath().getSegment()));
+		} else {
+
+			// DTO projections use merged metadata between domain type and result type
+			PersistentPropertyTranslator translator = PersistentPropertyTranslator.create(domainType,
+					Predicates.negate(CassandraPersistentProperty::hasExplicitColumnName));
+
+			CassandraPersistentEntity<?> persistentEntity = getQueryMapper().getConverter().getMappingContext()
+					.getRequiredPersistentEntity(projection.getMappedType());
+			for (CassandraPersistentProperty property : persistentEntity) {
+				columns.include(translator.translate(property).getColumnName());
+			}
+		}
+
 		Columns projectedColumns = Columns.empty();
 
 		if (returnType.isInterface()) {
 
-			ProjectionInformation projectionInformation = projectionFactory.getProjectionInformation(returnType);
+			ProjectionInformation projectionInformation = cassandraConverter.getProjectionFactory()
+					.getProjectionInformation(returnType);
 
 			if (projectionInformation.isClosed()) {
 
@@ -584,7 +600,7 @@ public class StatementFactory {
 				}
 			}
 		} else {
-			for (PersistentProperty<?> property : persistentEntity) {
+			for (PersistentProperty<?> property : domainType) {
 				projectedColumns = projectedColumns.include(property.getName());
 			}
 		}

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

@@ -31,6 +31,7 @@ import org.springframework.data.mapping.Association;
 import org.springframework.data.mapping.PersistentEntity;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.lang.Nullable;
+import org.springframework.util.ObjectUtils;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 
@@ -63,6 +64,14 @@ class AnnotatedCassandraConstructorProperty implements CassandraPersistentProper
 		return delegate.getColumnName();
 	}
 
+	@Override
+	public boolean hasExplicitColumnName() {
+		if (annotations.isPresent(Column.class)) {
+			return !ObjectUtils.isEmpty(annotations.get(Column.class).getString("value"));
+		}
+		return false;
+	}
+
 	@Override
 	@Nullable
 	public Integer getOrdinal() {

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

@@ -119,6 +119,11 @@ class CassandraConstructorProperty implements CassandraPersistentProperty {
 		return name;
 	}
 
+	@Override
+	public boolean hasExplicitColumnName() {
+		return false;
+	}
+
 	@Override
 	public Class<?> getType() {
 		return null;

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

@@ -21,10 +21,13 @@ import org.springframework.data.cassandra.core.mapping.CassandraPersistentProper
 import org.springframework.data.cassandra.core.mapping.MapId;
 import org.springframework.data.convert.CustomConversions;
 import org.springframework.data.convert.EntityConverter;
+import org.springframework.data.projection.EntityProjection;
+import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
+import com.datastax.oss.driver.api.core.cql.Row;
 import com.datastax.oss.driver.api.core.type.codec.registry.CodecRegistry;
 
 /**
@@ -38,9 +41,17 @@ public interface CassandraConverter
 		extends EntityConverter<CassandraPersistentEntity<?>, CassandraPersistentProperty, Object, Object> {
 
 	/**
-	 * Returns the {@link CustomConversions} registered in the {@link CassandraConverter}.
+	 * Returns the {@link ProjectionFactory} for this converter.
 	 *
-	 * @return the {@link CustomConversions}.
+	 * @return will never be {@literal null}.
+	 * @since 3.4
+	 */
+	ProjectionFactory getProjectionFactory();
+
+	/**
+	 * Returns the {@link CustomConversions} for this converter.
+	 *
+	 * @return will never be {@literal null}.
 	 */
 	CustomConversions getCustomConversions();
 
@@ -67,6 +78,19 @@ public interface CassandraConverter
 	 */
 	ColumnTypeResolver getColumnTypeResolver();
 
+	/**
+	 * Apply a projection to {@link Row} and return the projection return type {@code R}.
+	 * {@link EntityProjection#isProjection() Non-projecting} descriptors fall back to {@link #read(Class, Object) regular
+	 * object materialization}.
+	 *
+	 * @param descriptor the projection descriptor, must not be {@literal null}.
+	 * @param row must not be {@literal null}.
+	 * @param <R>
+	 * @return a new instance of the projection return type {@code R}.
+	 * @since 3.4
+	 */
+	<R> R project(EntityProjection<R, ?> descriptor, Row row);
+
 	/**
 	 * Returns the Id for an entity. It can return:
 	 * <ul>

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

@@ -18,10 +18,12 @@ package org.springframework.data.cassandra.core.convert;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.function.Function;
+import java.util.function.Predicate;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -37,8 +39,14 @@ import org.springframework.core.convert.support.DefaultConversionService;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.cassandra.core.mapping.*;
 import org.springframework.data.cassandra.core.mapping.Embedded.OnEmpty;
+import org.springframework.data.convert.CustomConversions;
+import org.springframework.data.mapping.AccessOptions;
 import org.springframework.data.mapping.MappingException;
+import org.springframework.data.mapping.PersistentEntity;
+import org.springframework.data.mapping.PersistentProperty;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
+import org.springframework.data.mapping.PersistentPropertyPath;
+import org.springframework.data.mapping.PersistentPropertyPathAccessor;
 import org.springframework.data.mapping.PreferredConstructor;
 import org.springframework.data.mapping.PreferredConstructor.Parameter;
 import org.springframework.data.mapping.context.MappingContext;
@@ -49,7 +57,11 @@ import org.springframework.data.mapping.model.ParameterValueProvider;
 import org.springframework.data.mapping.model.SpELContext;
 import org.springframework.data.mapping.model.SpELExpressionEvaluator;
 import org.springframework.data.mapping.model.SpELExpressionParameterValueProvider;
+import org.springframework.data.projection.EntityProjection;
+import org.springframework.data.projection.ProjectionFactory;
+import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 import org.springframework.data.util.ClassTypeInformation;
+import org.springframework.data.util.Predicates;
 import org.springframework.data.util.TypeInformation;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -95,6 +107,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 
 	private final DefaultColumnTypeResolver cassandraTypeResolver;
 	private final EmbeddedEntityOperations embeddedEntityOperations;
+	private final SpelAwareProxyProjectionFactory projectionFactory = new SpelAwareProxyProjectionFactory();
 
 	/**
 	 * Create a new {@link MappingCassandraConverter} with a {@link CassandraMappingContext}.
@@ -167,6 +180,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	@Override
 	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
 		this.spELContext = new SpELContext(this.spELContext, applicationContext);
+		this.projectionFactory.setBeanFactory(applicationContext);
 	}
 
 	/* (non-Javadoc)
@@ -175,12 +189,24 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	@Override
 	public void setBeanClassLoader(ClassLoader classLoader) {
 		this.beanClassLoader = classLoader;
+		this.projectionFactory.setBeanClassLoader(classLoader);
 	}
 
 	private TypeCodec<Object> getCodec(CassandraPersistentProperty property) {
 		return getCodecRegistry().codecFor(cassandraTypeResolver.resolve(property).getDataType());
 	}
 
+	/**
+	 * Returns the {@link ProjectionFactory} for this converter.
+	 *
+	 * @return will never be {@literal null}.
+	 * @since 3.4
+	 */
+	@Override
+	public ProjectionFactory getProjectionFactory() {
+		return projectionFactory;
+	}
+
 	/**
 	 * Sets the {@link CodecRegistry}.
 	 *
@@ -290,6 +316,119 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		}
 	}
 
+	@Override
+	public <R> R project(EntityProjection<R, ?> projection, Row row) {
+
+		if (!projection.isProjection()) {
+
+			TypeInformation<?> typeToRead = projection.getMappedType().getType().isInterface() ? projection.getDomainType()
+					: projection.getMappedType();
+			return (R) read(typeToRead.getType(), row);
+		}
+
+		ProjectingConversionContext context = new ProjectingConversionContext(getCustomConversions(), this::doReadRow,
+				this::doReadTupleValue, this::doReadUdtValue, this::readCollectionOrArray, this::readMap,
+				this::getPotentiallyConvertedSimpleRead, projection);
+
+		return doReadProjection(context, new RowValueProvider(row, new DefaultSpELExpressionEvaluator(row, spELContext)),
+				projection);
+	}
+
+	@SuppressWarnings("unchecked")
+	private <R> R doReadProjection(ConversionContext context, CassandraValueProvider valueProvider,
+			EntityProjection<R, ?> projection) {
+
+		CassandraPersistentEntity<?> entity = getMappingContext()
+				.getRequiredPersistentEntity(projection.getActualDomainType());
+		TypeInformation<?> mappedType = projection.getActualMappedType();
+		CassandraPersistentEntity<R> mappedEntity = (CassandraPersistentEntity<R>) getMappingContext()
+				.getPersistentEntity(mappedType);
+
+		boolean isInterfaceProjection = mappedType.getType().isInterface();
+		if (isInterfaceProjection) {
+
+			PersistentPropertyTranslator propertyTranslator = PersistentPropertyTranslator.create(mappedEntity);
+			PersistentPropertyAccessor<?> accessor = PropertyTranslatingPropertyAccessor
+					.create(new MapPersistentPropertyAccessor(), propertyTranslator);
+
+			readProperties(context, entity, valueProvider, accessor, Predicates.isTrue());
+			return (R) projectionFactory.createProjection(mappedType.getType(), accessor.getBean());
+		}
+
+		// DTO projection
+		if (mappedEntity == null) {
+			throw new MappingException(String.format("No mapping metadata found for %s", mappedType.getType().getName()));
+		}
+
+		// create target instance, merge metadata from underlying DTO type
+		PersistentPropertyTranslator propertyTranslator = PersistentPropertyTranslator.create(entity,
+				Predicates.negate(CassandraPersistentProperty::hasExplicitColumnName));
+		CassandraValueProvider valueProviderToUse = new TranslatingCassandraValueProvider(propertyTranslator,
+				valueProvider);
+
+		PreferredConstructor<?, CassandraPersistentProperty> persistenceConstructor = mappedEntity
+				.getPersistenceConstructor();
+
+		ParameterValueProvider<CassandraPersistentProperty> provider;
+		if (persistenceConstructor != null && persistenceConstructor.hasParameters()) {
+			SpELExpressionEvaluator evaluator = new DefaultSpELExpressionEvaluator(valueProviderToUse.getSource(),
+					spELContext);
+			ParameterValueProvider<CassandraPersistentProperty> parameterValueProvider = newParameterValueProvider(context,
+					entity, valueProviderToUse);
+			provider = new ConverterAwareSpELExpressionParameterValueProvider(evaluator, getConversionService(),
+					parameterValueProvider, context);
+		} else {
+			provider = NoOpParameterValueProvider.INSTANCE;
+		}
+
+		EntityInstantiator instantiator = instantiators.getInstantiatorFor(mappedEntity);
+		R instance = instantiator.createInstance(mappedEntity, provider);
+		PersistentPropertyAccessor<R> accessor = mappedEntity.getPropertyAccessor(instance);
+
+		readProperties(context, mappedEntity, valueProviderToUse, accessor, Predicates.isTrue());
+
+		return accessor.getBean();
+	}
+
+	private Object doReadOrProject(ConversionContext context, Row row, TypeInformation<?> typeHint,
+			EntityProjection<?, ?> typeDescriptor) {
+
+		if (typeDescriptor.isProjection()) {
+
+			CassandraValueProvider valueProvider = new RowValueProvider(row,
+					new DefaultSpELExpressionEvaluator(row, this.spELContext));
+			return doReadProjection(context, valueProvider, typeDescriptor);
+		}
+
+		return doReadRow(context, row, typeHint);
+	}
+
+	private Object doReadOrProject(ConversionContext context, UdtValue udtValue, TypeInformation<?> typeHint,
+			EntityProjection<?, ?> typeDescriptor) {
+
+		if (typeDescriptor.isProjection()) {
+
+			CassandraValueProvider valueProvider = new UdtValueProvider(udtValue,
+					new DefaultSpELExpressionEvaluator(udtValue, this.spELContext));
+			return doReadProjection(context, valueProvider, typeDescriptor);
+		}
+
+		return doReadUdtValue(context, udtValue, typeHint);
+	}
+
+	private Object doReadOrProject(ConversionContext context, TupleValue tupleValue, TypeInformation<?> typeHint,
+			EntityProjection<?, ?> typeDescriptor) {
+
+		if (typeDescriptor.isProjection()) {
+
+			CassandraValueProvider valueProvider = new TupleValueProvider(tupleValue,
+					new DefaultSpELExpressionEvaluator(tupleValue, this.spELContext));
+			return doReadProjection(context, valueProvider, typeDescriptor);
+		}
+
+		return doReadTupleValue(context, tupleValue, typeHint);
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.convert.EntityReader#read(java.lang.Class, S)
 	 */
@@ -419,7 +558,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		if (entity.requiresPropertyPopulation()) {
 			ConvertingPropertyAccessor<S> propertyAccessor = newConvertingPropertyAccessor(instance, entity);
 
-			readProperties(context, entity, valueProvider, propertyAccessor);
+			readProperties(context, entity, valueProvider, propertyAccessor, isConstructorArgument(entity).negate());
 			return propertyAccessor.getBean();
 		}
 
@@ -427,17 +566,19 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	}
 
 	private void readProperties(ConversionContext context, CassandraPersistentEntity<?> entity,
-			CassandraValueProvider valueProvider, PersistentPropertyAccessor<?> propertyAccessor) {
+			CassandraValueProvider valueProvider, PersistentPropertyAccessor<?> propertyAccessor,
+			Predicate<CassandraPersistentProperty> propertyFilter) {
 
 		for (CassandraPersistentProperty property : entity) {
 
-			// if true then skip; property was set in the constructor
-			if (entity.isConstructorArgument(property)) {
+			if (!propertyFilter.test(property)) {
 				continue;
 			}
 
+			ConversionContext contextToUse = context.forProperty(property.getName());
+
 			if (property.isCompositePrimaryKey() || valueProvider.hasProperty(property) || property.isEmbedded()) {
-				propertyAccessor.setProperty(property, getReadValue(context, valueProvider, property));
+				propertyAccessor.setProperty(property, getReadValue(contextToUse, valueProvider, property));
 			}
 		}
 	}
@@ -1109,6 +1250,10 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		return Map.class.isAssignableFrom(mapType) ? mapType : Map.class;
 	}
 
+	static Predicate<CassandraPersistentProperty> isConstructorArgument(PersistentEntity<?, ?> entity) {
+		return entity::isConstructorArgument;
+	}
+
 	enum NoOpParameterValueProvider implements ParameterValueProvider<CassandraPersistentProperty> {
 
 		INSTANCE;
@@ -1162,19 +1307,19 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	 */
 	protected static class ConversionContext {
 
-		private final org.springframework.data.convert.CustomConversions conversions;
+		final org.springframework.data.convert.CustomConversions conversions;
 
-		private final ContainerValueConverter<Row> rowConverter;
+		final ContainerValueConverter<Row> rowConverter;
 
-		private final ContainerValueConverter<TupleValue> tupleConverter;
+		final ContainerValueConverter<TupleValue> tupleConverter;
 
-		private final ContainerValueConverter<UdtValue> udtConverter;
+		final ContainerValueConverter<UdtValue> udtConverter;
 
-		private final ContainerValueConverter<Collection<?>> collectionConverter;
+		final ContainerValueConverter<Collection<?>> collectionConverter;
 
-		private final ContainerValueConverter<Map<?, ?>> mapConverter;
+		final ContainerValueConverter<Map<?, ?>> mapConverter;
 
-		private final ValueConverter<Object> elementConverter;
+		final ValueConverter<Object> elementConverter;
 
 		public ConversionContext(org.springframework.data.convert.CustomConversions conversions,
 				ContainerValueConverter<Row> rowConverter,
@@ -1190,6 +1335,10 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			this.elementConverter = elementConverter;
 		}
 
+		public ConversionContext forProperty(String name) {
+			return this;
+		}
+
 		/**
 		 * Converts a source object into {@link TypeInformation target}.
 		 *
@@ -1335,4 +1484,133 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		}
 	}
 
+	private static class PropertyTranslatingPropertyAccessor<T> implements PersistentPropertyPathAccessor<T> {
+
+		private final PersistentPropertyAccessor<T> delegate;
+		private final PersistentPropertyTranslator propertyTranslator;
+
+		private PropertyTranslatingPropertyAccessor(PersistentPropertyAccessor<T> delegate,
+				PersistentPropertyTranslator propertyTranslator) {
+			this.delegate = delegate;
+			this.propertyTranslator = propertyTranslator;
+		}
+
+		static <T> PersistentPropertyAccessor<T> create(PersistentPropertyAccessor<T> delegate,
+				PersistentPropertyTranslator propertyTranslator) {
+			return new PropertyTranslatingPropertyAccessor<>(delegate, propertyTranslator);
+		}
+
+		@Override
+		public void setProperty(PersistentProperty property, @Nullable Object value) {
+			delegate.setProperty(translate(property), value);
+		}
+
+		@Override
+		public Object getProperty(PersistentProperty<?> property) {
+			return delegate.getProperty(translate(property));
+		}
+
+		@Override
+		public T getBean() {
+			return delegate.getBean();
+		}
+
+		@Override
+		public void setProperty(PersistentPropertyPath<? extends PersistentProperty<?>> path, Object value,
+				AccessOptions.SetOptions options) {
+			throw new UnsupportedOperationException();
+		}
+
+		@Override
+		public Object getProperty(PersistentPropertyPath<? extends PersistentProperty<?>> path,
+				AccessOptions.GetOptions context) {
+			throw new UnsupportedOperationException();
+		}
+
+		@Override
+		public void setProperty(PersistentPropertyPath<? extends PersistentProperty<?>> path, Object value) {
+			throw new UnsupportedOperationException();
+		}
+
+		private CassandraPersistentProperty translate(PersistentProperty<?> property) {
+			return propertyTranslator.translate((CassandraPersistentProperty) property);
+		}
+	}
+
+	static class TranslatingCassandraValueProvider implements CassandraValueProvider {
+
+		private final PersistentPropertyTranslator translator;
+		private final CassandraValueProvider delegate;
+
+		public TranslatingCassandraValueProvider(PersistentPropertyTranslator translator, CassandraValueProvider delegate) {
+			this.translator = translator;
+			this.delegate = delegate;
+		}
+
+		@Override
+		public boolean hasProperty(CassandraPersistentProperty property) {
+			return delegate.hasProperty(translator.translate(property));
+		}
+
+		@Nullable
+		@Override
+		public <T> T getPropertyValue(CassandraPersistentProperty property) {
+			return delegate.getPropertyValue(translator.translate(property));
+		}
+
+		@Override
+		public Object getSource() {
+			return delegate.getSource();
+		}
+
+	}
+
+	class ProjectingConversionContext extends ConversionContext {
+
+		private final EntityProjection<?, ?> projection;
+
+		public ProjectingConversionContext(CustomConversions conversions, ContainerValueConverter<Row> rowConverter,
+				ContainerValueConverter<TupleValue> tupleConverter, ContainerValueConverter<UdtValue> udtConverter,
+				ContainerValueConverter<Collection<?>> collectionConverter, ContainerValueConverter<Map<?, ?>> mapConverter,
+				ValueConverter<Object> elementConverter, EntityProjection<?, ?> projection) {
+			super(conversions, (context, source, typeHint) -> doReadOrProject(context, source, typeHint, projection),
+					(context, source, typeHint) -> doReadOrProject(context, source, typeHint, projection),
+					(context, source, typeHint) -> doReadOrProject(context, source, typeHint, projection), collectionConverter,
+					mapConverter, elementConverter);
+			this.projection = projection;
+		}
+
+		@Override
+		public ConversionContext forProperty(String name) {
+
+			EntityProjection<?, ?> property = projection.findProperty(name);
+			if (property == null) {
+				return super.forProperty(name);
+			}
+
+			return new ProjectingConversionContext(conversions, rowConverter, tupleConverter, udtConverter,
+					collectionConverter, mapConverter, elementConverter, property);
+		}
+	}
+
+	static class MapPersistentPropertyAccessor implements PersistentPropertyAccessor<Map<String, Object>> {
+
+		Map<String, Object> map = new LinkedHashMap<>();
+
+		@Override
+		public void setProperty(PersistentProperty<?> persistentProperty, Object o) {
+			map.put(persistentProperty.getName(), o);
+		}
+
+		@Override
+		public Object getProperty(PersistentProperty<?> persistentProperty) {
+			return map.get(persistentProperty.getName());
+		}
+
+		@Override
+		public Map<String, Object> getBean() {
+			return map;
+		}
+	}
+
 }

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

@@ -42,6 +42,7 @@ import org.springframework.expression.spel.support.StandardEvaluationContext;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
+import org.springframework.util.ObjectUtils;
 import org.springframework.util.StringUtils;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
@@ -209,7 +210,6 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 				overriddenName = primaryKey.value();
 				forceQuote = primaryKey.forceQuote();
 			}
-
 		} else if (isPrimaryKeyColumn()) { // then it's a simple type
 
 			PrimaryKeyColumn primaryKeyColumn = findAnnotation(PrimaryKeyColumn.class);
@@ -218,7 +218,6 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 				overriddenName = primaryKeyColumn.value();
 				forceQuote = primaryKeyColumn.forceQuote();
 			}
-
 		} else { // then it's a vanilla column with the assumption that it's mapped to a single column
 
 			Column column = findAnnotation(Column.class);
@@ -232,6 +231,32 @@ public class BasicCassandraPersistentProperty extends AnnotationBasedPersistentP
 		return createColumnName(defaultName, overriddenName, forceQuote);
 	}
 
+	@Override
+	public boolean hasExplicitColumnName() {
+
+		if (isCompositePrimaryKey()) {
+			return false;
+		}
+
+		if (isIdProperty()) { // then the id is of a simple type (since it's not a composite primary key)
+
+			PrimaryKey primaryKey = findAnnotation(PrimaryKey.class);
+
+			return primaryKey != null && !ObjectUtils.isEmpty(primaryKey.value());
+
+		} else if (isPrimaryKeyColumn()) { // then it's a simple type
+
+			PrimaryKeyColumn primaryKeyColumn = findAnnotation(PrimaryKeyColumn.class);
+
+			return primaryKeyColumn != null && !ObjectUtils.isEmpty(primaryKeyColumn.value());
+		} else { // then it's a vanilla column with the assumption that it's mapped to a single column
+
+			Column column = findAnnotation(Column.class);
+
+			return column != null && !ObjectUtils.isEmpty(column.value());
+		}
+	}
+
 	@Nullable
 	private CqlIdentifier createColumnName(Supplier<String> defaultName, @Nullable String overriddenName,
 			boolean forceQuote) {

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

@@ -101,6 +101,14 @@ public interface CassandraPersistentProperty
 	@Deprecated
 	void setForceQuote(boolean forceQuote);
 
+	/**
+	 * Return whether the property has an explicitly configured column name.
+	 *
+	 * @return
+	 * @since 3.4
+	 */
+	boolean hasExplicitColumnName();
+
 	/**
 	 * The name of the element ordinal to which the property is persisted when the owning type is a mapped tuple.
 	 */

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

@@ -357,6 +357,11 @@ public class EmbeddedEntityOperations {
 			delegate.setForceQuote(forceQuote);
 		}
 
+		@Override
+		public boolean hasExplicitColumnName() {
+			return false;
+		}
+
 		@Override
 		@org.springframework.lang.Nullable
 		public Integer getOrdinal() {

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

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2021 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
+ *
+ *      https://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.function.Predicate;
+
+import org.springframework.data.util.Predicates;
+import org.springframework.lang.Nullable;
+
+/**
+ * Utility to translate a {@link CassandraPersistentProperty} into a corresponding property from a different
+ * {@link CassandraPersistentEntity} by looking it up by name.
+ * <p>
+ * Mainly used within the framework.
+ *
+ * @author Mark Paluch
+ * @since 3.4
+ */
+public class PersistentPropertyTranslator {
+
+	/**
+	 * Translate a {@link CassandraPersistentProperty} into a corresponding property from a different
+	 * {@link CassandraPersistentEntity}.
+	 *
+	 * @param property must not be {@literal null}.
+	 * @return the translated property. Can be the original {@code property}.
+	 */
+	public CassandraPersistentProperty translate(CassandraPersistentProperty property) {
+		return property;
+	}
+
+	/**
+	 * Create a new {@link PersistentPropertyTranslator}.
+	 *
+	 * @param targetEntity must not be {@literal null}.
+	 * @return the property translator to use.
+	 */
+	public static PersistentPropertyTranslator create(@Nullable CassandraPersistentEntity<?> targetEntity) {
+		return create(targetEntity, Predicates.isTrue());
+	}
+
+	/**
+	 * Create a new {@link PersistentPropertyTranslator} accepting a {@link Predicate filter predicate} whether the
+	 * translation should happen at all.
+	 *
+	 * @param targetEntity must not be {@literal null}.
+	 * @param translationFilter must not be {@literal null}.
+	 * @return the property translator to use.
+	 */
+	public static PersistentPropertyTranslator create(@Nullable CassandraPersistentEntity<?> targetEntity,
+			Predicate<CassandraPersistentProperty> translationFilter) {
+		return targetEntity != null ? new EntityPropertyTranslator(targetEntity, translationFilter)
+				: new PersistentPropertyTranslator();
+	}
+
+	private static class EntityPropertyTranslator extends PersistentPropertyTranslator {
+
+		private final CassandraPersistentEntity<?> targetEntity;
+		private final Predicate<CassandraPersistentProperty> translationFilter;
+
+		EntityPropertyTranslator(CassandraPersistentEntity<?> targetEntity,
+				Predicate<CassandraPersistentProperty> translationFilter) {
+			this.targetEntity = targetEntity;
+			this.translationFilter = translationFilter;
+		}
+
+		@Override
+		public CassandraPersistentProperty translate(CassandraPersistentProperty property) {
+
+			if (!translationFilter.test(property)) {
+				return property;
+			}
+
+			CassandraPersistentProperty targetProperty = targetEntity.getPersistentProperty(property.getName());
+			return targetProperty != null ? targetProperty : property;
+		}
+	}
+
+}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/CassandraRepositoryFactory.java

@@ -18,6 +18,7 @@ package org.springframework.data.cassandra.repository.support;
 import java.lang.reflect.Method;
 import java.util.Optional;
 
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.data.cassandra.core.CassandraOperations;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
@@ -71,6 +72,11 @@ public class CassandraRepositoryFactory extends RepositoryFactorySupport {
 		this.mappingContext = operations.getConverter().getMappingContext();
 	}
 
+	@Override
+	protected ProjectionFactory getProjectionFactory(ClassLoader classLoader, BeanFactory beanFactory) {
+		return this.operations.getConverter().getProjectionFactory();
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.repository.core.support.RepositoryFactorySupport#getRepositoryBaseClass(org.springframework.data.repository.core.RepositoryMetadata)
 	 */

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactory.java

@@ -18,6 +18,7 @@ package org.springframework.data.cassandra.repository.support;
 import java.lang.reflect.Method;
 import java.util.Optional;
 
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.data.cassandra.core.ReactiveCassandraOperations;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.CassandraPersistentProperty;
@@ -70,6 +71,11 @@ public class ReactiveCassandraRepositoryFactory extends ReactiveRepositoryFactor
 		setEvaluationContextProvider(ReactiveQueryMethodEvaluationContextProvider.DEFAULT);
 	}
 
+	@Override
+	protected ProjectionFactory getProjectionFactory(ClassLoader classLoader, BeanFactory beanFactory) {
+		return this.operations.getConverter().getProjectionFactory();
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.repository.core.support.RepositoryFactorySupport#getRepositoryBaseClass(org.springframework.data.repository.core.RepositoryMetadata)
 	 */

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

@@ -20,6 +20,7 @@ import static org.springframework.data.cassandra.core.mapping.BasicMapId.*;
 import static org.springframework.data.cassandra.test.util.RowMockUtil.*;
 
 import lombok.AllArgsConstructor;
+import lombok.Data;
 import lombok.EqualsAndHashCode;
 import lombok.NoArgsConstructor;
 import lombok.RequiredArgsConstructor;
@@ -60,6 +61,9 @@ import org.springframework.data.cassandra.domain.TypeWithMapId;
 import org.springframework.data.cassandra.domain.User;
 import org.springframework.data.cassandra.domain.UserToken;
 import org.springframework.data.cassandra.test.util.RowMockUtil;
+import org.springframework.data.projection.EntityProjection;
+import org.springframework.data.projection.EntityProjectionIntrospector;
+import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.cql.Row;
@@ -1046,6 +1050,32 @@ public class MappingCassandraConverterUnitTests {
 		assertThat(converted.tuple.firstname).isEqualTo("Two");
 	}
 
+	@Test
+	void shouldConsiderNestedProjections() {
+
+		DefaultTupleValue value = new DefaultTupleValue(
+				new DefaultTupleType(Arrays.asList(DataTypes.ASCII, DataTypes.ASCII, DataTypes.ASCII)));
+
+		value.setString(0, "Zero");
+		value.setString(1, "One");
+		value.setString(2, "Two");
+
+		EntityProjectionIntrospector introspector = EntityProjectionIntrospector.create(
+				new SpelAwareProxyProjectionFactory(), EntityProjectionIntrospector.ProjectionPredicate.typeHierarchy(),
+				this.mappingContext);
+
+		rowMock = RowMockUtil.newRowMock(RowMockUtil.column("firstname", "Heisenberg", DataTypes.ASCII),
+				RowMockUtil.column("tuple", value, value.getType()));
+
+		EntityProjection<WithMappedTupleDtoProjection, WithMappedTuple> projection = introspector
+				.introspect(WithMappedTupleDtoProjection.class, WithMappedTuple.class);
+
+		WithMappedTupleDtoProjection result = this.mappingCassandraConverter.project(projection, rowMock);
+
+		assertThat(result.getFirstname()).isEqualTo("Heisenberg");
+		assertThat(result.getTuple().getOne()).isEqualTo("One");
+	}
+
 	private static List<Object> getValues(Map<CqlIdentifier, Object> statement) {
 		return new ArrayList<>(statement.values());
 	}
@@ -1312,6 +1342,20 @@ public class MappingCassandraConverterUnitTests {
 		TupleWithElementAnnotationInConstructor tuple;
 	}
 
+	@Data
+	private static class WithMappedTupleDtoProjection {
+
+		String firstname;
+		TupleProjection tuple;
+	}
+
+	private static interface TupleProjection {
+
+		String getZero();
+
+		String getOne();
+	}
+
 	@Tuple
 	private static class TupleWithElementAnnotationInConstructor {
 

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/CassandraRepositoryFactoryUnitTests.java

@@ -25,6 +25,8 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
 
 import org.springframework.data.cassandra.core.CassandraTemplate;
 import org.springframework.data.cassandra.core.convert.CassandraConverter;
@@ -32,6 +34,7 @@ import org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentE
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.domain.Person;
 import org.springframework.data.cassandra.repository.query.CassandraEntityInformation;
+import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 import org.springframework.data.repository.Repository;
 
 /**
@@ -40,6 +43,7 @@ import org.springframework.data.repository.Repository;
  * @author Mark Paluch
  */
 @ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
 @SuppressWarnings({ "rawtypes", "unchecked" })
 public class CassandraRepositoryFactoryUnitTests {
 
@@ -52,6 +56,7 @@ public class CassandraRepositoryFactoryUnitTests {
 	void setUp() {
 
 		when(template.getConverter()).thenReturn(converter);
+		when(converter.getProjectionFactory()).thenReturn(new SpelAwareProxyProjectionFactory());
 		when(converter.getMappingContext()).thenReturn(mappingContext);
 	}
 

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/ReactiveCassandraRepositoryFactoryUnitTests.java

@@ -25,6 +25,8 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
 
 import org.springframework.data.cassandra.core.ReactiveCassandraTemplate;
 import org.springframework.data.cassandra.core.convert.CassandraConverter;
@@ -32,6 +34,7 @@ import org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentE
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
 import org.springframework.data.cassandra.domain.Person;
 import org.springframework.data.cassandra.repository.query.CassandraEntityInformation;
+import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
 import org.springframework.data.repository.Repository;
 
 /**
@@ -40,6 +43,7 @@ import org.springframework.data.repository.Repository;
  * @author Mark Paluch
  */
 @ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
 @SuppressWarnings({ "rawtypes", "unchecked" })
 public class ReactiveCassandraRepositoryFactoryUnitTests {
 
@@ -52,6 +56,7 @@ public class ReactiveCassandraRepositoryFactoryUnitTests {
 	void setUp() {
 
 		when(template.getConverter()).thenReturn(converter);
+		when(converter.getProjectionFactory()).thenReturn(new SpelAwareProxyProjectionFactory());
 		when(converter.getMappingContext()).thenReturn(mappingContext);
 	}