Add support for QueryResultConverter.

Closes #1568

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

@@ -128,7 +128,7 @@ public interface AsyncCassandraOperations {
 
 	/**
 	 * Execute a {@code SELECT} query with paging and convert the result set to a {@link Slice} of entities. A sliced
-	 * query translates the effective {@link Statement#getFetchSize() fetch size} to the page size.
+	 * query translates the effective {@link Statement#getPageSize() fetch size} to the page size.
 	 *
 	 * @param statement the CQL statement, must not be {@literal null}.
 	 * @param entityClass The entity type must not be {@literal null}.

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

@@ -162,7 +162,7 @@ public interface CassandraOperations extends FluentCassandraOperations {
 
 	/**
 	 * Execute a {@code SELECT} query with paging and convert the result set to a {@link Slice} of entities. A sliced
-	 * query translates the effective {@link Statement#getFetchSize() fetch size} to the page size.
+	 * query translates the effective {@link Statement#getPageSize()} to the page size.
 	 *
 	 * @param statement the CQL statement, must not be {@literal null}.
 	 * @param entityClass The entity type must not be {@literal null}.

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

@@ -58,6 +58,7 @@ 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.Lazy;
 import org.springframework.util.Assert;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
@@ -353,10 +354,17 @@ 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(EntityProjection.nonProjecting(entityClass),
-				EntityQueryUtils.getTableName(statement));
+		return doSelect(statement, entityClass, getTableName(entityClass), entityClass, QueryResultConverter.entity());
+	}
 
-		return doQuery(statement, (row, rowNum) -> mapper.apply(row));
+	<T, R> List<R> doSelect(Statement<?> statement, Class<?> entityClass, CqlIdentifier tableName, Class<T> returnType,
+			QueryResultConverter<T, R> mappingFunction) {
+
+		EntityProjection<T, ?> projection = entityOperations.introspectProjection(returnType, entityClass);
+
+		RowMapper<R> rowMapper = getRowMapper(projection, tableName, mappingFunction);
+
+		return doQuery(statement, rowMapper);
 	}
 
 	@Override
@@ -372,13 +380,14 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		Assert.notNull(statement, "Statement must not be null");
 		Assert.notNull(entityClass, "Entity type must not be null");
 
+		return doSlice(statement,
+				getRowMapper(entityClass, EntityQueryUtils.getTableName(statement), QueryResultConverter.entity()));
+	}
+
+	<T> Slice<T> doSlice(Statement<?> statement, RowMapper<T> mapper) {
+
 		ResultSet resultSet = doQueryForResultSet(statement);
-
-		Function<Row, T> mapper = getMapper(EntityProjection.nonProjecting(entityClass),
-				EntityQueryUtils.getTableName(statement));
-
-		return EntityQueryUtils.readSlice(resultSet, (row, rowNum) -> mapper.apply(row), 0,
-				getEffectivePageSize(statement));
+		return EntityQueryUtils.readSlice(resultSet, mapper, 0, getEffectivePageSize(statement));
 	}
 
 	@Override
@@ -387,9 +396,17 @@ 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(EntityProjection.nonProjecting(entityClass),
-				EntityQueryUtils.getTableName(statement));
-		return doQueryForStream(statement, (row, rowNum) -> mapper.apply(row));
+		return doStream(statement, entityClass, EntityQueryUtils.getTableName(statement), entityClass,
+				QueryResultConverter.entity());
+	}
+
+	<T, R> Stream<R> doStream(Statement<?> statement, Class<?> entityClass, CqlIdentifier tableName, Class<T> returnType,
+			QueryResultConverter<T, R> mappingFunction) {
+
+		EntityProjection<T, ?> projection = entityOperations.introspectProjection(returnType, entityClass);
+
+		RowMapper<R> rowMapper = getRowMapper(projection, tableName, mappingFunction);
+		return doQueryForStream(statement, rowMapper);
 	}
 
 	// -------------------------------------------------------------------------
@@ -402,10 +419,11 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		Assert.notNull(query, "Query must not be null");
 		Assert.notNull(entityClass, "Entity type must not be null");
 
-		return doSelect(query, entityClass, getTableName(entityClass), entityClass);
+		return doSelect(query, entityClass, getTableName(entityClass), entityClass, QueryResultConverter.entity());
 	}
 
-	<T> List<T> doSelect(Query query, Class<?> entityClass, CqlIdentifier tableName, Class<T> returnType) {
+	<T, R> List<R> doSelect(Query query, Class<?> entityClass, CqlIdentifier tableName, Class<T> returnType,
+			QueryResultConverter<? super T, ? extends R> mappingFunction) {
 
 		CassandraPersistentEntity<?> entity = getRequiredPersistentEntity(entityClass);
 		EntityProjection<T, ?> projection = entityOperations.introspectProjection(returnType, entityClass);
@@ -415,9 +433,9 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		Query queryToUse = query.columns(columns);
 
 		StatementBuilder<Select> select = getStatementFactory().select(queryToUse, entity, tableName);
-		Function<Row, T> mapper = getMapper(projection, tableName);
+		RowMapper<R> rowMapper = getRowMapper(projection, tableName, mappingFunction);
 
-		return doQuery(select.build(), (row, rowNum) -> mapper.apply(row));
+		return doQuery(select.build(), rowMapper);
 	}
 
 	@Override
@@ -434,9 +452,24 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		Assert.notNull(query, "Query must not be null");
 		Assert.notNull(entityClass, "Entity type must not be null");
 
-		StatementBuilder<Select> select = getStatementFactory().select(query, getRequiredPersistentEntity(entityClass));
+		return doSlice(query, entityClass, getRequiredPersistentEntity(entityClass).getTableName(), entityClass,
+				QueryResultConverter.entity());
+	}
 
-		return slice(select.build(), entityClass);
+	<T, R> Slice<R> doSlice(Query query, Class<?> entityClass, CqlIdentifier tableName, Class<T> returnType,
+			QueryResultConverter<? super T, ? extends R> mappingFunction) {
+
+		CassandraPersistentEntity<?> entity = getRequiredPersistentEntity(entityClass);
+		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);
+		RowMapper<R> rowMapper = getRowMapper(projection, tableName, mappingFunction);
+
+		return doSlice(select.build(), rowMapper);
 	}
 
 	@Override
@@ -445,17 +478,19 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		Assert.notNull(query, "Query must not be null");
 		Assert.notNull(entityClass, "Entity type must not be null");
 
-		return doStream(query, entityClass, getTableName(entityClass), entityClass);
+		return doStream(query, entityClass, getTableName(entityClass), entityClass, QueryResultConverter.entity());
 	}
 
-	<T> Stream<T> doStream(Query query, Class<?> entityClass, CqlIdentifier tableName, Class<T> returnType) {
+	<T, R> Stream<R> doStream(Query query, Class<?> entityClass, CqlIdentifier tableName, Class<T> returnType,
+			QueryResultConverter<? super T, ? extends R> mappingFunction) {
 
 		StatementBuilder<Select> select = getStatementFactory().select(query, getRequiredPersistentEntity(entityClass),
 				tableName);
 		EntityProjection<T, ?> projection = entityOperations.introspectProjection(returnType, entityClass);
 
-		Function<Row, T> mapper = getMapper(projection, tableName);
-		return doQueryForStream(select.build(), (row, rowNum) -> mapper.apply(row));
+		RowMapper<R> rowMapper = getRowMapper(projection, tableName, mappingFunction);
+
+		return doQueryForStream(select.build(), rowMapper);
 	}
 
 	@Override
@@ -767,6 +802,16 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		return new ExecutableSelectOperationSupport(this).query(domainType);
 	}
 
+	@Override
+	public UntypedSelect query(String cql) {
+		return new ExecutableSelectOperationSupport(this).query(cql);
+	}
+
+	@Override
+	public UntypedSelect query(Statement<?> statement) {
+		return new ExecutableSelectOperationSupport(this).query(statement);
+	}
+
 	@Override
 	public <T> ExecutableInsert<T> insert(Class<T> domainType) {
 		return new ExecutableInsertOperationSupport(this).insert(domainType);
@@ -909,6 +954,32 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		return getCqlOperations().execute(new GetConfiguredPageSize());
 	}
 
+	@SuppressWarnings("unchecked")
+	<T, R> RowMapper<R> getRowMapper(EntityProjection<T, ?> projection, CqlIdentifier tableName,
+			QueryResultConverter<? super T, ? extends R> mappingFunction) {
+
+		Function<Row, T> mapper = getMapper(projection, tableName);
+
+		return mappingFunction == QueryResultConverter.entity() ? (row, rowNum) -> (R) mapper.apply(row)
+				: (row, rowNum) -> {
+					Lazy<T> reader = Lazy.of(() -> mapper.apply(row));
+					return mappingFunction.mapRow(row, reader::get);
+				};
+	}
+
+	@SuppressWarnings("unchecked")
+	<T, R> RowMapper<R> getRowMapper(Class<T> domainClass, CqlIdentifier tableName,
+			QueryResultConverter<? super T, ? extends R> mappingFunction) {
+
+		Function<Row, T> mapper = getMapper(EntityProjection.nonProjecting(domainClass), tableName);
+
+		return mappingFunction == QueryResultConverter.entity() ? (row, rowNum) -> (R) mapper.apply(row)
+				: (row, rowNum) -> {
+					Lazy<T> reader = Lazy.of(() -> mapper.apply(row));
+					return mappingFunction.mapRow(row, reader::get);
+				};
+	}
+
 	@SuppressWarnings("unchecked")
 	private <T> Function<Row, T> getMapper(EntityProjection<T, ?> projection, CqlIdentifier tableName) {
 

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

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2025 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;
+
+import com.datastax.oss.driver.api.core.cql.Row;
+
+enum EntityResultConverter implements QueryResultConverter<Object, Object> {
+
+	INSTANCE;
+
+	@Override
+	public Object mapRow(Row row, ConversionResultSupplier<Object> reader) {
+		return reader.get();
+	}
+
+	@Override
+	public <V> QueryResultConverter<Object, V> andThen(QueryResultConverter<? super Object, ? extends V> after) {
+		return (QueryResultConverter) after;
+	}
+}

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

@@ -17,15 +17,21 @@ package org.springframework.data.cassandra.core;
 
 import java.util.List;
 import java.util.Optional;
+import java.util.function.Function;
 import java.util.stream.Stream;
 
 import org.jspecify.annotations.Nullable;
+
+import org.springframework.data.cassandra.core.cql.RowMapper;
 import org.springframework.data.cassandra.core.query.Query;
+import org.springframework.data.domain.Slice;
 import org.springframework.lang.Contract;
 import org.springframework.util.Assert;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
 import com.datastax.oss.driver.api.core.cql.ResultSet;
+import com.datastax.oss.driver.api.core.cql.Row;
+import com.datastax.oss.driver.api.core.cql.Statement;
 
 /**
  * The {@link ExecutableSelectOperation} interface allows creation and execution of Cassandra {@code SELECT} operations
@@ -68,6 +74,76 @@ public interface ExecutableSelectOperation {
 	 */
 	<T> ExecutableSelect<T> query(Class<T> domainType);
 
+	/**
+	 * Begin creating a Cassandra {@code SELECT} query operation for the given {@code cql}. The given {@code cql} must be
+	 * a {@code SELECT} query.
+	 *
+	 * @param cql {@code SELECT} statement, must not be {@literal null}.
+	 * @return new instance of {@link UntypedSelect}.
+	 * @throws IllegalArgumentException if {@code cql} is {@literal null}.
+	 * @since 5.0
+	 * @see ExecutableSelect
+	 */
+	UntypedSelect query(String cql);
+
+	/**
+	 * Begin creating a Cassandra {@code SELECT} query operation for the given {@link Statement}. The given
+	 * {@link Statement} must be a {@code SELECT} query.
+	 *
+	 * @param statement {@code SELECT} statement, must not be {@literal null}.
+	 * @return new instance of {@link UntypedSelect}.
+	 * @throws IllegalArgumentException if {@link Statement statement} is {@literal null}.
+	 * @since 5.0
+	 * @see ExecutableSelect
+	 */
+	UntypedSelect query(Statement<?> statement);
+
+	/**
+	 * Select query that is not yet associated with a result type.
+	 *
+	 * @since 5.0
+	 */
+	interface UntypedSelect {
+
+		/**
+		 * Define the {@link Class result target type} that the Cassandra Row fields should be mapped to.
+		 *
+		 * @param resultType result type; must not be {@literal null}.
+		 * @param <T> {@link Class type} of the result.
+		 * @return new instance of {@link TerminatingResults}.
+		 * @throws IllegalArgumentException if {@link Class resultType} is {@literal null}.
+		 */
+		@Contract("_ -> new")
+		<T> TerminatingResults<T> as(Class<T> resultType);
+
+		/**
+		 * Configure a {@link Function mapping function} that maps the Cassandra Row to a result type. This is a simplified
+		 * variant of {@link #map(RowMapper)}.
+		 *
+		 * @param mapper row mapping function; must not be {@literal null}.
+		 * @param <T> {@link Class type} of the result.
+		 * @return new instance of {@link TerminatingResults}.
+		 * @throws IllegalArgumentException if {@link Function mapper} is {@literal null}.
+		 * @see #map(RowMapper)
+		 */
+		@Contract("_ -> new")
+		default <T> TerminatingResults<T> map(Function<Row, ? extends T> mapper) {
+			return map((row, rowNum) -> mapper.apply(row));
+		}
+
+		/**
+		 * Configure a {@link RowMapper} that maps the Cassandra Row to a result type.
+		 *
+		 * @param mapper the row mapper; must not be {@literal null}.
+		 * @param <T> {@link Class type} of the result.
+		 * @return new instance of {@link TerminatingResults}.
+		 * @throws IllegalArgumentException if {@link RowMapper mapper} is {@literal null}.
+		 */
+		@Contract("_ -> new")
+		<T> TerminatingResults<T> map(RowMapper<T> mapper);
+
+	}
+
 	/**
 	 * Table override (optional).
 	 */
@@ -121,7 +197,7 @@ public interface ExecutableSelectOperation {
 		 * @param <R> {@link Class type} of the result.
 		 * @param resultType desired {@link Class target type} of the result; must not be {@literal null}.
 		 * @return new instance of {@link SelectWithQuery}.
-		 * @throws IllegalArgumentException if resultType is {@literal null}.
+		 * @throws IllegalArgumentException if {@link Class resultType} is {@literal null}.
 		 * @see SelectWithQuery
 		 */
 		@Contract("_ -> new")
@@ -130,18 +206,19 @@ public interface ExecutableSelectOperation {
 	}
 
 	/**
-	 * Filtering (optional).
+	 * Define a {@link Query} used as the filter for the {@code SELECT}.
 	 */
 	interface SelectWithQuery<T> extends TerminatingSelect<T> {
 
 		/**
-		 * Set the {@link Query} to use as a filter.
+		 * Set the {@link Query} used as a filter in the {@code SELECT} statement.
 		 *
 		 * @param query {@link Query} used as a filter; must not be {@literal null}.
 		 * @return new instance of {@link TerminatingSelect}.
 		 * @throws IllegalArgumentException if {@link Query} is {@literal null}.
 		 * @see TerminatingSelect
 		 */
+		@Contract("_ -> new")
 		TerminatingSelect<T> matching(Query query);
 
 	}
@@ -149,7 +226,13 @@ public interface ExecutableSelectOperation {
 	/**
 	 * Trigger {@code SELECT} query execution by calling one of the terminating methods.
 	 */
-	interface TerminatingSelect<T> {
+	interface TerminatingSelect<T> extends TerminatingProjections, TerminatingResults<T> {}
+
+	/**
+	 * Trigger {@code SELECT} query execution by calling one of the terminating methods returning result projections for
+	 * count and exists projections.
+	 */
+	interface TerminatingProjections {
 
 		/**
 		 * Get the number of matching elements.
@@ -168,6 +251,25 @@ public interface ExecutableSelectOperation {
 			return count() > 0;
 		}
 
+	}
+
+	/**
+	 * Trigger {@code SELECT} query execution by calling one of the terminating methods and return mapped results.
+	 */
+	interface TerminatingResults<T> {
+
+		/**
+		 * Map the query result to a different type using {@link QueryResultConverter}.
+		 *
+		 * @param <R> {@link Class type} of the result.
+		 * @param converter the converter, must not be {@literal null}.
+		 * @return new instance of {@link TerminatingResults}.
+		 * @throws IllegalArgumentException if {@link QueryResultConverter converter} is {@literal null}.
+		 * @since 5.0
+		 */
+		@Contract("_ -> new")
+		<R> TerminatingResults<R> map(QueryResultConverter<? super T, ? extends R> converter);
+
 		/**
 		 * Get the first result, or no result.
 		 *
@@ -214,6 +316,15 @@ public interface ExecutableSelectOperation {
 		 */
 		List<T> all();
 
+		/**
+		 * Execute the query with paging and convert the result set to a {@link Slice} of entities. A sliced query
+		 * translates the effective {@link Statement#getPageSize() fetch size} to the page size.
+		 *
+		 * @return the converted results
+		 * @since 5.0
+		 */
+		Slice<T> slice();
+
 		/**
 		 * Stream all matching elements.
 		 *

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

@@ -19,12 +19,18 @@ import java.util.List;
 import java.util.stream.Stream;
 
 import org.jspecify.annotations.Nullable;
+
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
+import org.springframework.data.cassandra.core.cql.QueryExtractorDelegate;
+import org.springframework.data.cassandra.core.cql.RowMapper;
 import org.springframework.data.cassandra.core.query.Query;
+import org.springframework.data.domain.Slice;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
+import com.datastax.oss.driver.api.core.cql.Statement;
 
 /**
  * Implementation of {@link ExecutableSelectOperation}.
@@ -47,36 +53,182 @@ class ExecutableSelectOperationSupport implements ExecutableSelectOperation {
 
 		Assert.notNull(domainType, "DomainType must not be null");
 
-		return new ExecutableSelectSupport<>(this.template, domainType, domainType, Query.empty(), null);
+		return new ExecutableSelectSupport<>(this.template, domainType, domainType, QueryResultConverter.entity(),
+				Query.empty(), null);
 	}
 
-	static class ExecutableSelectSupport<T> implements ExecutableSelect<T> {
+	@Override
+	public UntypedSelect query(String cql) {
+
+		Assert.hasText(cql, "CQL must not be empty");
+
+		return new UntypedSelectSupport(this.template, SimpleStatement.newInstance(cql));
+	}
+
+	@Override
+	public UntypedSelect query(Statement<?> statement) {
+
+		Assert.notNull(statement, "Statement must not be null");
+
+		return new UntypedSelectSupport(this.template, statement);
+	}
+
+	private record UntypedSelectSupport(CassandraTemplate template, Statement<?> statement) implements UntypedSelect {
+
+		@Override
+		public <T> TerminatingResults<T> as(Class<T> resultType) {
+
+			Assert.notNull(resultType, "Result type must not be null");
+
+			return new TypedSelectSupport<>(template, statement, resultType);
+		}
+
+		@Override
+		public <T> TerminatingResults<T> map(RowMapper<T> mapper) {
+
+			Assert.notNull(mapper, "RowMapper must not be null");
+
+			return new TerminatingSelectResultSupport<>(template, statement, mapper);
+		}
+
+	}
+
+	static class TypedSelectSupport<T> extends TerminatingSelectResultSupport<T, T> implements TerminatingResults<T> {
+
+		private final Class<T> domainType;
+
+		TypedSelectSupport(CassandraTemplate template, Statement<?> statement, Class<T> domainType) {
+			super(template, statement,
+					template.getRowMapper(domainType, EntityQueryUtils.getTableName(statement), QueryResultConverter.entity()));
+
+			this.domainType = domainType;
+		}
+
+		@Override
+		public <R> TerminatingResults<R> map(QueryResultConverter<? super T, ? extends R> converter) {
+
+			Assert.notNull(converter, "Mapping function must not be null");
+
+			return new TerminatingSelectResultSupport<>(this.template, this.statement, this.domainType, converter);
+		}
+
+	}
+
+	static class TerminatingSelectResultSupport<S, T> implements TerminatingResults<T> {
+
+		final CassandraTemplate template;
+
+		final Statement<?> statement;
+
+		final RowMapper<T> rowMapper;
+
+		TerminatingSelectResultSupport(CassandraTemplate template, Statement<?> statement, RowMapper<T> rowMapper) {
+			this.template = template;
+			this.statement = statement;
+			this.rowMapper = rowMapper;
+		}
+
+		TerminatingSelectResultSupport(CassandraTemplate template, Statement<?> statement, Class<S> domainType,
+				QueryResultConverter<? super S, ? extends T> mappingFunction) {
+			this(template, statement,
+					template.getRowMapper(domainType, EntityQueryUtils.getTableName(statement), mappingFunction));
+		}
+
+		@Override
+		public <R> TerminatingResults<R> map(QueryResultConverter<? super T, ? extends R> converter) {
+
+			return new TerminatingSelectResultSupport<>(this.template, this.statement, (row, rowNum) -> {
+
+				return converter.mapRow(row, () -> {
+					return this.rowMapper.mapRow(row, rowNum);
+				});
+			});
+		}
+
+		@Override
+		public @Nullable T firstValue() {
+
+			List<T> result = this.template.getCqlOperations().query(this.statement, this.rowMapper);
+
+			return ObjectUtils.isEmpty(result) ? null : result.iterator().next();
+		}
+
+		@Override
+		public @Nullable T oneValue() {
+
+			List<T> result = this.template.getCqlOperations().query(this.statement, this.rowMapper);
+
+			if (ObjectUtils.isEmpty(result)) {
+				return null;
+			}
+
+			if (result.size() > 1) {
+				throw new IncorrectResultSizeDataAccessException(
+						String.format("Query [%s] returned non unique result", QueryExtractorDelegate.getCql(this.statement)), 1);
+			}
+
+			return result.iterator().next();
+		}
+
+		@Override
+		public List<T> all() {
+			return this.template.getCqlOperations().query(this.statement, this.rowMapper);
+		}
+
+		@Override
+		public Slice<T> slice() {
+			return this.template.doSlice(this.statement, this.rowMapper);
+		}
+
+		@Override
+		public Stream<T> stream() {
+			return this.template.getCqlOperations().queryForStream(this.statement, this.rowMapper);
+		}
+
+	}
+
+	static class ExecutableSelectSupport<S, T> implements ExecutableSelect<T> {
 
 		private final CassandraTemplate template;
 
 		private final Class<?> domainType;
 
-		private final Class<T> returnType;
+		private final Class<S> returnType;
+
+		private final QueryResultConverter<? super S, ? extends T> mappingFunction;
 
 		private final Query query;
 
 		private final @Nullable CqlIdentifier tableName;
 
-		public ExecutableSelectSupport(CassandraTemplate template, Class<?> domainType, Class<T> returnType, Query query,
+		public ExecutableSelectSupport(CassandraTemplate template, Class<?> domainType, Class<S> returnType,
+				QueryResultConverter<? super S, ? extends T> mappingFunction, Query query,
 				@Nullable CqlIdentifier tableName) {
+
 			this.template = template;
 			this.domainType = domainType;
 			this.returnType = returnType;
+			this.mappingFunction = mappingFunction;
 			this.query = query;
 			this.tableName = tableName;
 		}
 
+		@Override
+		public <R> TerminatingResults<R> map(QueryResultConverter<? super T, ? extends R> converter) {
+
+			Assert.notNull(converter, "Mapping function name must not be null");
+
+			return new ExecutableSelectSupport<>(this.template, this.domainType, this.returnType,
+					this.mappingFunction.andThen(converter), this.query, tableName);
+		}
+
 		@Override
 		public SelectWithProjection<T> inTable(CqlIdentifier tableName) {
 
 			Assert.notNull(tableName, "Table name must not be null");
 
-			return new ExecutableSelectSupport<>(this.template, this.domainType, this.returnType, this.query, tableName);
+			return new ExecutableSelectSupport<>(this.template, this.domainType, this.returnType, this.mappingFunction,
+					this.query, tableName);
 		}
 
 		@Override
@@ -84,7 +236,8 @@ class ExecutableSelectOperationSupport implements ExecutableSelectOperation {
 
 			Assert.notNull(returnType, "ReturnType must not be null");
 
-			return new ExecutableSelectSupport<>(this.template, this.domainType, returnType, this.query, this.tableName);
+			return new ExecutableSelectSupport<>(this.template, this.domainType, returnType, QueryResultConverter.entity(),
+					this.query, this.tableName);
 		}
 
 		@Override
@@ -92,7 +245,8 @@ class ExecutableSelectOperationSupport implements ExecutableSelectOperation {
 
 			Assert.notNull(query, "Query must not be null");
 
-			return new ExecutableSelectSupport<>(this.template, this.domainType, this.returnType, query, this.tableName);
+			return new ExecutableSelectSupport<>(this.template, this.domainType, this.returnType, this.mappingFunction, query,
+					this.tableName);
 		}
 
 		@Override
@@ -108,7 +262,8 @@ class ExecutableSelectOperationSupport implements ExecutableSelectOperation {
 		@Override
 		public @Nullable T firstValue() {
 
-			List<T> result = this.template.doSelect(this.query.limit(1), this.domainType, getTableName(), this.returnType);
+			List<T> result = this.template.doSelect(this.query.limit(1), this.domainType, getTableName(), this.returnType,
+					this.mappingFunction);
 
 			return ObjectUtils.isEmpty(result) ? null : result.iterator().next();
 		}
@@ -116,7 +271,8 @@ class ExecutableSelectOperationSupport implements ExecutableSelectOperation {
 		@Override
 		public @Nullable T oneValue() {
 
-			List<T> result = this.template.doSelect(this.query.limit(2), this.domainType, getTableName(), this.returnType);
+			List<T> result = this.template.doSelect(this.query.limit(2), this.domainType, getTableName(), this.returnType,
+					this.mappingFunction);
 
 			if (ObjectUtils.isEmpty(result)) {
 				return null;
@@ -132,12 +288,17 @@ class ExecutableSelectOperationSupport implements ExecutableSelectOperation {
 
 		@Override
 		public List<T> all() {
-			return this.template.doSelect(this.query, this.domainType, getTableName(), this.returnType);
+			return this.template.doSelect(this.query, this.domainType, getTableName(), this.returnType, this.mappingFunction);
+		}
+
+		@Override
+		public Slice<T> slice() {
+			return this.template.doSlice(this.query, this.domainType, getTableName(), this.returnType, this.mappingFunction);
 		}
 
 		@Override
 		public Stream<T> stream() {
-			return this.template.doStream(this.query, this.domainType, getTableName(), this.returnType);
+			return this.template.doStream(this.query, this.domainType, getTableName(), this.returnType, this.mappingFunction);
 		}
 
 		private CqlIdentifier getTableName() {
@@ -146,4 +307,5 @@ class ExecutableSelectOperationSupport implements ExecutableSelectOperation {
 
 	}
 
+
 }

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

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2025 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;
+
+import com.datastax.oss.driver.api.core.cql.Row;
+
+/**
+ * Converter for Cassandra query results.
+ * <p>
+ * This is a functional interface that allows for mapping a {@link Row} to a result type.
+ * {@link #mapRow(Row, ConversionResultSupplier) row mapping} can obtain upstream a {@link ConversionResultSupplier
+ * upstream converter} to enrich the final result object. This is useful when e.g. wrapping result objects where the
+ * wrapper needs to obtain information from the actual {@link Row}.
+ *
+ * @param <T> object type accepted by this converter.
+ * @param <R> the returned result type.
+ * @author Mark Paluch
+ * @since 5.0
+ */
+@FunctionalInterface
+public interface QueryResultConverter<T, R> {
+
+	/**
+	 * Returns a function that returns the materialized entity.
+	 *
+	 * @param <T> the type of the input and output entity to the function.
+	 * @return a function that returns the materialized entity.
+	 */
+	@SuppressWarnings("unchecked")
+	static <T> QueryResultConverter<T, T> entity() {
+		return (QueryResultConverter<T, T>) EntityResultConverter.INSTANCE;
+	}
+
+	/**
+	 * Map a {@link Row} that is read from the Cassandra database to a query result.
+	 *
+	 * @param row the raw row from the Cassandra result.
+	 * @param reader reader object that supplies an upstream result from an earlier converter.
+	 * @return the mapped result.
+	 */
+	R mapRow(Row row, ConversionResultSupplier<T> reader);
+
+	/**
+	 * Returns a composed function that first applies this function to its input, and then applies the {@code after}
+	 * function to the result. If evaluation of either function throws an exception, it is relayed to the caller of the
+	 * composed function.
+	 *
+	 * @param <V> the type of output of the {@code after} function, and of the composed function.
+	 * @param after the function to apply after this function is applied.
+	 * @return a composed function that first applies this function and then applies the {@code after} function.
+	 */
+	default <V> QueryResultConverter<T, V> andThen(QueryResultConverter<? super R, ? extends V> after) {
+		return (row, reader) -> after.mapRow(row, () -> mapRow(row, reader));
+	}
+
+	/**
+	 * A supplier that converts a {@link Row} into {@code T}. Allows for lazy reading of query results.
+	 *
+	 * @param <T> type of the returned result.
+	 */
+	interface ConversionResultSupplier<T> {
+
+		/**
+		 * Obtain the upstream conversion result.
+		 *
+		 * @return the upstream conversion result.
+		 */
+		T get();
+
+	}
+
+}

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

@@ -129,7 +129,7 @@ public interface ReactiveCassandraOperations extends ReactiveFluentCassandraOper
 
 	/**
 	 * Execute a {@code SELECT} query with paging and convert the result set to a {@link Slice} of entities. A sliced
-	 * query translates the effective {@link Statement#getFetchSize() fetch size} to the page size.
+	 * query translates the effective {@link Statement#getPageSize() fetch size} to the page size.
 	 *
 	 * @param statement the CQL statement, must not be {@literal null}.
 	 * @param entityClass The entity type must not be {@literal null}.

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

@@ -66,6 +66,7 @@ import org.springframework.data.mapping.callback.ReactiveEntityCallbacks;
 import org.springframework.data.projection.EntityProjection;
 import org.springframework.data.projection.ProjectionFactory;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
+import org.springframework.data.util.Lazy;
 import org.springframework.util.Assert;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
@@ -391,10 +392,11 @@ public class ReactiveCassandraTemplate
 		Assert.notNull(query, "Query must not be null");
 		Assert.notNull(entityClass, "Entity type must not be null");
 
-		return doSelect(query, entityClass, getTableName(entityClass), entityClass);
+		return doSelect(query, entityClass, getTableName(entityClass), entityClass, QueryResultConverter.entity());
 	}
 
-	<T> Flux<T> doSelect(Query query, Class<?> entityClass, CqlIdentifier tableName, Class<T> returnType) {
+	<T, R> Flux<R> doSelect(Query query, Class<?> entityClass, CqlIdentifier tableName, Class<T> returnType,
+			QueryResultConverter<? super T, ? extends R> mappingFunction) {
 
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entityClass);
 		EntityProjection<T, ?> projection = entityOperations.introspectProjection(returnType, entityClass);
@@ -404,9 +406,9 @@ public class ReactiveCassandraTemplate
 		Query queryToUse = query.columns(columns);
 
 		StatementBuilder<Select> select = getStatementFactory().select(queryToUse, persistentEntity, tableName);
-		Function<Row, T> mapper = getMapper(projection, tableName);
+		RowMapper<R> mapper = getRowMapper(projection, tableName, mappingFunction);
 
-		return doQuery(select.build(), (row, rowNum) -> mapper.apply(row));
+		return doQuery(select.build(), mapper);
 	}
 
 	@Override
@@ -760,6 +762,16 @@ public class ReactiveCassandraTemplate
 		return new ReactiveSelectOperationSupport(this).query(domainType);
 	}
 
+	@Override
+	public UntypedSelect query(String cql) {
+		return new ReactiveSelectOperationSupport(this).query(cql);
+	}
+
+	@Override
+	public UntypedSelect query(Statement<?> statement) {
+		return new ReactiveSelectOperationSupport(this).query(statement);
+	}
+
 	@Override
 	public ReactiveUpdate update(Class<?> domainType) {
 		return new ReactiveUpdateOperationSupport(this).update(domainType);
@@ -881,6 +893,32 @@ public class ReactiveCassandraTemplate
 		return getReactiveCqlOperations().execute(new GetConfiguredPageSize()).single();
 	}
 
+	@SuppressWarnings("unchecked")
+	<T, R> RowMapper<R> getRowMapper(EntityProjection<T, ?> projection, CqlIdentifier tableName,
+			QueryResultConverter<? super T, ? extends R> mappingFunction) {
+
+		Function<Row, T> mapper = getMapper(projection, tableName);
+
+		return mappingFunction == QueryResultConverter.entity() ? (row, rowNum) -> (R) mapper.apply(row)
+				: (row, rowNum) -> {
+					Lazy<T> reader = Lazy.of(() -> mapper.apply(row));
+					return mappingFunction.mapRow(row, reader::get);
+				};
+	}
+
+	@SuppressWarnings("unchecked")
+	<T, R> RowMapper<R> getRowMapper(Class<T> domainClass, CqlIdentifier tableName,
+			QueryResultConverter<? super T, ? extends R> mappingFunction) {
+
+		Function<Row, T> mapper = getMapper(EntityProjection.nonProjecting(domainClass), tableName);
+
+		return mappingFunction == QueryResultConverter.entity() ? (row, rowNum) -> (R) mapper.apply(row)
+				: (row, rowNum) -> {
+					Lazy<T> reader = Lazy.of(() -> mapper.apply(row));
+					return mappingFunction.mapRow(row, reader::get);
+				};
+	}
+
 	@SuppressWarnings("unchecked")
 	private <T> Function<Row, T> getMapper(EntityProjection<T, ?> projection, CqlIdentifier tableName) {
 

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

@@ -18,11 +18,16 @@ package org.springframework.data.cassandra.core;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.util.function.Function;
+
+import org.springframework.data.cassandra.core.cql.RowMapper;
 import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.lang.Contract;
 import org.springframework.util.Assert;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
+import com.datastax.oss.driver.api.core.cql.Row;
+import com.datastax.oss.driver.api.core.cql.Statement;
 
 /**
  * The {@link ReactiveSelectOperation} interface allows creation and execution of Cassandra {@code SELECT} operations in
@@ -65,13 +70,83 @@ public interface ReactiveSelectOperation {
 	 */
 	<T> ReactiveSelect<T> query(Class<T> domainType);
 
+	/**
+	 * Begin creating a Cassandra {@code SELECT} query operation for the given {@code cql}. The given {@code cql} must be
+	 * a {@code SELECT} query.
+	 *
+	 * @param cql {@code SELECT} statement, must not be {@literal null}.
+	 * @return new instance of {@link UntypedSelect}.
+	 * @throws IllegalArgumentException if {@code cql} is {@literal null}.
+	 * @since 5.0
+	 * @see ReactiveSelect
+	 */
+	UntypedSelect query(String cql);
+
+	/**
+	 * Begin creating a Cassandra {@code SELECT} query operation for the given {@link Statement}. The given
+	 * {@link Statement} must be a {@code SELECT} query.
+	 *
+	 * @param statement {@code SELECT} statement, must not be {@literal null}.
+	 * @return new instance of {@link UntypedSelect}.
+	 * @throws IllegalArgumentException if {@link Statement statement} is {@literal null}.
+	 * @since 5.0
+	 * @see ReactiveSelect
+	 */
+	UntypedSelect query(Statement<?> statement);
+
+	/**
+	 * Select query that is not yet associated with a result type.
+	 *
+	 * @since 5.0
+	 */
+	interface UntypedSelect {
+
+		/**
+		 * Define the {@link Class result target type} that the Cassandra Row fields should be mapped to.
+		 *
+		 * @param resultType result type; must not be {@literal null}.
+		 * @param <T> {@link Class type} of the result.
+		 * @return new instance of {@link TerminatingResults}.
+		 * @throws IllegalArgumentException if {@link Class resultType} is {@literal null}.
+		 */
+		@Contract("_ -> new")
+		<T> TerminatingResults<T> as(Class<T> resultType);
+
+		/**
+		 * Configure a {@link Function mapping function} that maps the Cassandra Row to a result type. This is a simplified
+		 * variant of {@link #map(RowMapper)}.
+		 *
+		 * @param mapper row mapping function; must not be {@literal null}.
+		 * @param <T> {@link Class type} of the result.
+		 * @return new instance of {@link TerminatingResults}.
+		 * @throws IllegalArgumentException if {@link Function mapper} is {@literal null}.
+		 * @see #map(RowMapper)
+		 */
+		@Contract("_ -> new")
+		default <T> TerminatingResults<T> map(Function<Row, ? extends T> mapper) {
+			return map((row, rowNum) -> mapper.apply(row));
+		}
+
+		/**
+		 * Configure a {@link RowMapper} that maps the Cassandra Row to a result type.
+		 *
+		 * @param mapper the row mapper; must not be {@literal null}.
+		 * @param <T> {@link Class type} of the result.
+		 * @return new instance of {@link TerminatingResults}.
+		 * @throws IllegalArgumentException if {@link RowMapper mapper} is {@literal null}.
+		 */
+		@Contract("_ -> new")
+		<T> TerminatingResults<T> map(RowMapper<T> mapper);
+
+	}
+
 	/**
 	 * Table override (optional).
 	 */
 	interface SelectWithTable<T> extends SelectWithQuery<T> {
 
 		/**
-		 * Explicitly set the {@link String name} of the table on which to perform the query.
+		 * Explicitly set the {@link String name} of the table on which to execute the query.
 		 * <p>
 		 * Skip this step to use the default table derived from the {@link Class domain type}.
 		 *
@@ -90,7 +165,7 @@ public interface ReactiveSelectOperation {
 		}
 
 		/**
-		 * Explicitly set the {@link CqlIdentifier name} of the table on which to perform the query.
+		 * Explicitly set the {@link CqlIdentifier name} of the table on which to execute the query.
 		 * <p>
 		 * Skip this step to use the default table derived from the {@link Class domain type}.
 		 *
@@ -111,12 +186,12 @@ public interface ReactiveSelectOperation {
 	interface SelectWithProjection<T> extends SelectWithQuery<T> {
 
 		/**
-		 * Define the {@link Class result target type} that the fields should be mapped to.
+		 * Define the {@link Class result target type} that the Cassandra Row fields should be mapped to.
 		 * <p>
-		 * Skip this step if you are only interested in the original {@link Class domain type}.
+		 * Skip this step if you are anyway only interested in the original {@link Class domain type}.
 		 *
 		 * @param <R> {@link Class type} of the result.
-		 * @param resultType desired {@link Class type} of the result; must not be {@literal null}.
+		 * @param resultType desired {@link Class target type} of the result; must not be {@literal null}.
 		 * @return new instance of {@link SelectWithQuery}.
 		 * @throws IllegalArgumentException if {@link Class resultType} is {@literal null}.
 		 * @see SelectWithQuery
@@ -137,7 +212,6 @@ public interface ReactiveSelectOperation {
 		 * @param query {@link Query} used as a filter; must not be {@literal null}.
 		 * @return new instance of {@link TerminatingSelect}.
 		 * @throws IllegalArgumentException if {@link Query} is {@literal null}.
-		 * @see org.springframework.data.cassandra.core.query.Query
 		 * @see TerminatingSelect
 		 */
 		@Contract("_ -> new")
@@ -146,9 +220,15 @@ public interface ReactiveSelectOperation {
 	}
 
 	/**
-	 * Trigger {@code SELECT} execution by calling one of the terminating methods.
+	 * Trigger {@code SELECT} query execution by calling one of the terminating methods.
 	 */
-	interface TerminatingSelect<T> {
+	interface TerminatingSelect<T> extends TerminatingProjections, TerminatingResults<T> {}
+
+	/**
+	 * Trigger {@code SELECT} query execution by calling one of the terminating methods returning result projections for
+	 * count and exists projections.
+	 */
+	interface TerminatingProjections {
 
 		/**
 		 * Get the number of matching elements.
@@ -166,6 +246,25 @@ public interface ReactiveSelectOperation {
 		 */
 		Mono<Boolean> exists();
 
+	}
+
+	/**
+	 * Trigger {@code SELECT} execution by calling one of the terminating methods.
+	 */
+	interface TerminatingResults<T> {
+
+		/**
+		 * Map the query result to a different type using {@link QueryResultConverter}.
+		 *
+		 * @param <R> {@link Class type} of the result.
+		 * @param converter the converter, must not be {@literal null}.
+		 * @return new instance of {@link TerminatingResults}.
+		 * @throws IllegalArgumentException if {@link QueryResultConverter converter} is {@literal null}.
+		 * @since 5.0
+		 */
+		@Contract("_ -> new")
+		<R> TerminatingResults<R> map(QueryResultConverter<? super T, ? extends R> converter);
+
 		/**
 		 * Get the first result or no result.
 		 *

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

@@ -17,13 +17,22 @@ package org.springframework.data.cassandra.core;
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.publisher.SynchronousSink;
+
+import java.util.List;
+import java.util.function.Supplier;
 
 import org.jspecify.annotations.Nullable;
+
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
+import org.springframework.data.cassandra.core.cql.QueryExtractorDelegate;
+import org.springframework.data.cassandra.core.cql.RowMapper;
 import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.util.Assert;
 
 import com.datastax.oss.driver.api.core.CqlIdentifier;
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
+import com.datastax.oss.driver.api.core.cql.Statement;
 
 /**
  * Implementation of {@link ReactiveSelectOperation}.
@@ -46,36 +55,160 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
 
 		Assert.notNull(domainType, "DomainType must not be null");
 
-		return new ReactiveSelectSupport<>(this.template, domainType, domainType, Query.empty(), null);
+		return new ReactiveSelectSupport<>(this.template, domainType, domainType, QueryResultConverter.entity(),
+				Query.empty(), null);
 	}
 
-	static class ReactiveSelectSupport<T> implements ReactiveSelect<T> {
+	@Override
+	public UntypedSelect query(String cql) {
+
+		Assert.hasText(cql, "CQL must not be empty");
+
+		return new UntypedSelectSupport(this.template, SimpleStatement.newInstance(cql));
+	}
+
+	@Override
+	public UntypedSelect query(Statement<?> statement) {
+
+		Assert.notNull(statement, "Statement must not be null");
+
+		return new UntypedSelectSupport(this.template, statement);
+	}
+
+	private record UntypedSelectSupport(ReactiveCassandraTemplate template,
+			Statement<?> statement) implements UntypedSelect {
+
+		@Override
+		public <T> TerminatingResults<T> as(Class<T> resultType) {
+
+			Assert.notNull(resultType, "Result type must not be null");
+
+			return new TypedSelectSupport<>(template, statement, resultType);
+		}
+
+		@Override
+		public <T> TerminatingResults<T> map(RowMapper<T> mapper) {
+
+			Assert.notNull(mapper, "RowMapper must not be null");
+
+			return new TerminatingSelectResultSupport<>(template, statement, mapper);
+		}
+
+	}
+
+	static class TypedSelectSupport<T> extends TerminatingSelectResultSupport<T, T> implements TerminatingResults<T> {
+
+		private final Class<T> domainType;
+
+		TypedSelectSupport(ReactiveCassandraTemplate template, Statement<?> statement, Class<T> domainType) {
+			super(template, statement,
+					template.getRowMapper(domainType, EntityQueryUtils.getTableName(statement), QueryResultConverter.entity()));
+
+			this.domainType = domainType;
+		}
+
+		@Override
+		public <R> TerminatingResults<R> map(QueryResultConverter<? super T, ? extends R> converter) {
+
+			Assert.notNull(converter, "Mapping function must not be null");
+
+			return new TerminatingSelectResultSupport<>(this.template, this.statement, this.domainType, converter);
+		}
+
+	}
+
+	static class TerminatingSelectResultSupport<S, T> implements TerminatingResults<T> {
+
+		final ReactiveCassandraTemplate template;
+
+		final Statement<?> statement;
+
+		final RowMapper<T> rowMapper;
+
+		TerminatingSelectResultSupport(ReactiveCassandraTemplate template, Statement<?> statement, RowMapper<T> rowMapper) {
+			this.template = template;
+			this.statement = statement;
+			this.rowMapper = rowMapper;
+		}
+
+		TerminatingSelectResultSupport(ReactiveCassandraTemplate template, Statement<?> statement, Class<S> domainType,
+				QueryResultConverter<? super S, ? extends T> mappingFunction) {
+			this(template, statement,
+					template.getRowMapper(domainType, EntityQueryUtils.getTableName(statement), mappingFunction));
+		}
+
+		@Override
+		public <R> TerminatingResults<R> map(QueryResultConverter<? super T, ? extends R> converter) {
+
+			return new TerminatingSelectResultSupport<>(this.template, this.statement, (row, rowNum) -> {
+
+				return converter.mapRow(row, () -> {
+					return this.rowMapper.mapRow(row, rowNum);
+				});
+			});
+		}
+
+		@Override
+		public Mono<T> first() {
+			return this.template.getReactiveCqlOperations().query(this.statement, this.rowMapper).next();
+		}
+
+		@Override
+		public Mono<T> one() {
+
+			Flux<T> result = this.template.getReactiveCqlOperations().query(this.statement, this.rowMapper);
+
+			return result.collectList() //
+					.handle((objects, sink) -> handleOne(objects, sink, () -> QueryExtractorDelegate.getCql(this.statement)));
+		}
+
+		@Override
+		public Flux<T> all() {
+			return this.template.getReactiveCqlOperations().query(this.statement, this.rowMapper);
+		}
+
+	}
+
+	static class ReactiveSelectSupport<S, T> implements ReactiveSelect<T> {
 
 		private final ReactiveCassandraTemplate template;
 
 		private final Class<?> domainType;
 
-		private final Class<T> returnType;
+		private final Class<S> returnType;
+
+		private final QueryResultConverter<? super S, ? extends T> mappingFunction;
 
 		private final Query query;
 
 		private final @Nullable CqlIdentifier tableName;
 
-		public ReactiveSelectSupport(ReactiveCassandraTemplate template, Class<?> domainType, Class<T> returnType,
-				Query query, @Nullable CqlIdentifier tableName) {
+		public ReactiveSelectSupport(ReactiveCassandraTemplate template, Class<?> domainType, Class<S> returnType,
+				QueryResultConverter<? super S, ? extends T> mappingFunction, Query query, @Nullable CqlIdentifier tableName) {
 			this.template = template;
 			this.domainType = domainType;
 			this.returnType = returnType;
+			this.mappingFunction = mappingFunction;
 			this.query = query;
 			this.tableName = tableName;
 		}
 
+		@Override
+		public <R> TerminatingResults<R> map(QueryResultConverter<? super T, ? extends R> converter) {
+
+			Assert.notNull(converter, "Mapping function name must not be null");
+
+			return new ReactiveSelectSupport<>(this.template, this.domainType, this.returnType,
+					this.mappingFunction.andThen(converter), this.query, tableName);
+		}
+
 		@Override
 		public SelectWithProjection<T> inTable(CqlIdentifier tableName) {
 
 			Assert.notNull(tableName, "Table name must not be null");
 
-			return new ReactiveSelectSupport<>(this.template, this.domainType, this.returnType, this.query, tableName);
+			return new ReactiveSelectSupport<>(this.template, this.domainType, this.returnType, this.mappingFunction,
+					this.query, tableName);
 		}
 
 		@Override
@@ -83,7 +216,8 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
 
 			Assert.notNull(returnType, "ReturnType must not be null");
 
-			return new ReactiveSelectSupport<>(this.template, this.domainType, returnType, this.query, this.tableName);
+			return new ReactiveSelectSupport<>(this.template, this.domainType, returnType, QueryResultConverter.entity(),
+					this.query, this.tableName);
 		}
 
 		@Override
@@ -91,7 +225,8 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
 
 			Assert.notNull(query, "Query must not be null");
 
-			return new ReactiveSelectSupport<>(this.template, this.domainType, this.returnType, query, this.tableName);
+			return new ReactiveSelectSupport<>(this.template, this.domainType, this.returnType, this.mappingFunction, query,
+					this.tableName);
 		}
 
 		@Override
@@ -106,34 +241,26 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
 
 		@Override
 		public Mono<T> first() {
-			return this.template.doSelect(this.query.limit(1), this.domainType, getTableName(), this.returnType).next();
+
+			Flux<T> one = this.template.doSelect(this.query.limit(1), this.domainType, getTableName(), this.returnType,
+					this.mappingFunction);
+
+			return one.next();
 		}
 
 		@Override
 		public Mono<T> one() {
 
-			Flux<T> result = this.template.doSelect(this.query.limit(2), this.domainType, getTableName(), this.returnType);
+			Flux<T> result = this.template.doSelect(this.query.limit(2), this.domainType, getTableName(), this.returnType,
+					this.mappingFunction);
 
 			return result.collectList() //
-					.handle((objects, sink) -> {
-
-						if (objects.isEmpty()) {
-							return;
-						}
-
-						if (objects.size() == 1) {
-							sink.next(objects.get(0));
-							return;
-						}
-
-						sink.error(new IncorrectResultSizeDataAccessException(
-								String.format("Query [%s] returned non unique result", this.query), 1));
-					});
+					.handle((objects, sink) -> handleOne(objects, sink, this.query::toString));
 		}
 
 		@Override
 		public Flux<T> all() {
-			return this.template.doSelect(this.query, this.domainType, getTableName(), this.returnType);
+			return this.template.doSelect(this.query, this.domainType, getTableName(), this.returnType, this.mappingFunction);
 		}
 
 		private CqlIdentifier getTableName() {
@@ -142,4 +269,19 @@ class ReactiveSelectOperationSupport implements ReactiveSelectOperation {
 
 	}
 
+	private static <T> void handleOne(List<T> objects, SynchronousSink<T> sink, Supplier<@Nullable String> query) {
+
+		if (objects.isEmpty()) {
+			return;
+		}
+
+		if (objects.size() == 1) {
+			sink.next(objects.get(0));
+			return;
+		}
+
+		sink.error(new IncorrectResultSizeDataAccessException(
+				String.format("Query [%s] returned non unique result", query.get()), 1));
+	}
+
 }

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/query/CassandraQueryExecution.java

@@ -164,6 +164,44 @@ interface CassandraQueryExecution {
 
 	}
 
+	final class SearchExecution implements CassandraQueryExecution {
+
+		private final CassandraOperations operations;
+		private final CassandraParameterAccessor accessor;
+
+		public SearchExecution(CassandraOperations operations, CassandraParameterAccessor accessor) {
+
+			this.operations = operations;
+			this.accessor = accessor;
+		}
+
+		@Override
+		public Object execute(Statement<?> statement, Class<?> type) {
+
+			ScoringFunction function = accessor.getScoringFunction();
+
+			List<SearchResult<?>> results = operations.select(statement, type, (o, row) -> {
+
+				if (row.getColumnDefinitions().contains("__score__")) {
+					return new SearchResult<>(o, getScore(row, "__score__", function));
+				}
+
+				if (row.getColumnDefinitions().contains("score")) {
+					return new SearchResult<>(o, getScore(row, "score", function));
+				}
+				return new SearchResult<>(o, 0);
+			});
+
+			return new SearchResults(results);
+		}
+
+		private Score getScore(Row row, String columnName, ScoringFunction function) {
+
+			Object object = row.getObject(columnName);
+			return Score.of(((Number) object).doubleValue(), function);
+		}
+	}
+
 	/**
 	 * {@link CassandraQueryExecution} to return a single entity.
 	 *

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

@@ -306,8 +306,7 @@ class CassandraTemplateUnitTests {
 		assertThat(beforeSave).isSameAs(user);
 	}
 
-	@Test
-		// GH-1295
+	@Test // GH-1295
 	void insertShouldConsiderEntityAfterCallback() {
 
 		when(resultSet.wasApplied()).thenReturn(true);
@@ -325,8 +324,7 @@ class CassandraTemplateUnitTests {
 				.isEqualTo("INSERT INTO users (id,firstname,lastname) VALUES ('ww','Walter','White')");
 	}
 
-	@Test
-		// DATACASS-618
+	@Test // DATACASS-618
 	void insertShouldInsertVersionedEntity() {
 
 		when(resultSet.wasApplied()).thenReturn(true);
@@ -342,8 +340,7 @@ class CassandraTemplateUnitTests {
 		assertThat(beforeSave).isSameAs(user);
 	}
 
-	@Test
-		// GH-1295
+	@Test // GH-1295
 	void insertShouldInsertVersionedEntityAfterCallback() {
 
 		when(resultSet.wasApplied()).thenReturn(true);
@@ -361,8 +358,7 @@ class CassandraTemplateUnitTests {
 				"INSERT INTO vusers (id,version,firstname,lastname) VALUES ('ww',0,'Walter','White') IF NOT EXISTS");
 	}
 
-	@Test
-		// DATACASS-250
+	@Test // DATACASS-250
 	void insertShouldInsertWithOptionsEntity() {
 
 		InsertOptions insertOptions = InsertOptions.builder().withIfNotExists().build();

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

@@ -19,10 +19,13 @@ import static org.assertj.core.api.Assertions.*;
 import static org.springframework.data.cassandra.core.query.Criteria.*;
 import static org.springframework.data.cassandra.core.query.Query.*;
 
+import java.util.List;
+import java.util.Optional;
 import java.util.stream.Stream;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.data.annotation.Id;
@@ -34,6 +37,7 @@ import org.springframework.data.cassandra.core.mapping.Table;
 import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.data.cassandra.repository.support.SchemaTestUtils;
 import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTests;
+import org.springframework.data.domain.SearchResult;
 import org.springframework.util.ObjectUtils;
 
 /**
@@ -81,7 +85,7 @@ class ExecutableSelectOperationSupportIntegrationTests extends AbstractKeyspaceC
 
 	@Test // DATACASS-485
 	void domainTypeIsRequired() {
-		assertThatIllegalArgumentException().isThrownBy(() -> this.template.query(null));
+		assertThatIllegalArgumentException().isThrownBy(() -> this.template.query((Class<? extends Object>) null));
 	}
 
 	@Test // DATACASS-485
@@ -137,6 +141,27 @@ class ExecutableSelectOperationSupportIntegrationTests extends AbstractKeyspaceC
 		assertThat(this.template.query(Person.class).as(Jedi.class).all()).hasOnlyElementsOfType(Jedi.class).isNotEmpty();
 	}
 
+	@Test // GH-1568
+	void findAllByWithResultConverter() {
+
+		List<SearchResult<Jedi>> results = this.template.query(Person.class).as(Jedi.class)
+				.map((row, reader) -> new SearchResult<>(reader.get(), 0)).all();
+
+		assertThat(results).hasOnlyElementsOfType(SearchResult.class).isNotEmpty();
+		assertThat(results).extracting(SearchResult::getContent).hasOnlyElementsOfType(Jedi.class);
+	}
+
+	@Test // GH-1568
+	void findAllByWithStagedResultConverter() {
+
+		List<Optional<SearchResult<Jedi>>> results = this.template.query(Person.class).as(Jedi.class)
+				.map((row, reader) -> new SearchResult<>(reader.get(), 0)).map((row, reader) -> Optional.of(reader.get()))
+				.all();
+
+		assertThat(results).hasOnlyElementsOfType(Optional.class).isNotEmpty();
+		assertThat(results).extracting(Optional::get).hasOnlyElementsOfType(SearchResult.class);
+	}
+
 	@Test // DATACASS-485
 	void findBy() {
 		assertThat(this.template.query(Person.class).matching(queryLuke()).one()).contains(luke);

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

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2025 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;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import org.junit.jupiter.api.Test;
+
+import com.datastax.oss.driver.api.core.cql.Row;
+
+/**
+ * Unit tests for {@link QueryResultConverter}.
+ *
+ * @author Mark Paluch
+ */
+class QueryResultConverterUnitTests {
+
+	static final QueryResultConverter.ConversionResultSupplier<Row> ERROR_SUPPLIER = () -> {
+		throw new IllegalStateException("must not read conversion result");
+	};
+
+	@Test // GH-1568
+	void converterDoesNotEagerlyRetrieveConversionResultFromSupplier() {
+
+		QueryResultConverter<Row, String> converter = (row, reader) -> "done";
+
+		assertThat(converter.mapRow(mock(Row.class), ERROR_SUPPLIER)).isEqualTo("done");
+	}
+
+	@Test // GH-1568
+	void converterPassesOnConversionResultToNextStage() {
+
+		Row source = mock(Row.class);
+
+		QueryResultConverter<Row, Integer> stagedConverter = ((QueryResultConverter<Row, Integer>) (row, reader) -> 1)
+				.andThen((row, reader) -> {
+
+					assertThat(row).isEqualTo(source);
+					return Integer.valueOf(reader.get());
+				});
+
+		assertThat(stagedConverter.mapRow(source, ERROR_SUPPLIER)).isEqualTo(1);
+	}
+
+	@Test // GH-1568
+	void entityConverterDelaysConversion() {
+
+		Row source = mock(Row.class);
+
+		QueryResultConverter<Row, Integer> converter = QueryResultConverter.<Row> entity().andThen((row, reader) -> {
+
+			assertThat(row).isEqualTo(source);
+			return Integer.valueOf(20);
+		});
+
+		assertThat(converter.mapRow(source, ERROR_SUPPLIER)).isEqualTo(20);
+	}
+
+}

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

@@ -23,8 +23,11 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
+import java.util.Optional;
+
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.dao.IncorrectResultSizeDataAccessException;
 import org.springframework.data.annotation.Id;
@@ -84,7 +87,7 @@ class ReactiveSelectOperationSupportIntegrationTests extends AbstractKeyspaceCre
 
 	@Test // DATACASS-485
 	void domainTypeIsRequired() {
-		assertThatIllegalArgumentException().isThrownBy(() -> this.template.query(null));
+		assertThatIllegalArgumentException().isThrownBy(() -> this.template.query((Class<? extends Object>) null));
 	}
 
 	@Test // DATACASS-485
@@ -167,6 +170,18 @@ class ReactiveSelectOperationSupportIntegrationTests extends AbstractKeyspaceCre
 				.assertNext(actual -> assertThat(actual).isNotEmpty().hasOnlyElementsOfType(Jedi.class)).verifyComplete();
 	}
 
+	@Test // GH-1568
+	void findAllByWithResultConverter() {
+
+		Flux<Optional<Jedi>> result = this.template.query(Person.class).as(Jedi.class)
+				.map((row, reader) -> Optional.of(reader.get())).all();
+
+		result.collectList().as(StepVerifier::create)
+				.assertNext(
+						actual -> assertThat(actual).extracting(Optional::get).isNotEmpty().hasOnlyElementsOfType(Jedi.class))
+				.verifyComplete();
+	}
+
 	@Test // DATACASS-485
 	void findBy() {