Add support for QueryResultConverter.

Closes #1568

Author: mp911de

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

@@ -128,7 +128,7 @@ <T> CompletableFuture<Void> select(String cql, Consumer<T> entityConsumer, Class
 
 	/**
 	 * 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 @@ default CassandraBatchOperations batchOps() {
 
 	/**
 	 * 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.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 <T> List<T> select(Statement<?> statement, Class<T> entityClass) {
 		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());
+	}
+
+	<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, (row, rowNum) -> mapper.apply(row));
+		return doQuery(statement, rowMapper);
 	}
 
 	@Override
@@ -372,13 +380,14 @@ public <T> Slice<T> slice(Statement<?> statement, Class<T> entityClass) {
 		Assert.notNull(statement, "Statement must not be null");
 		Assert.notNull(entityClass, "Entity type must not be null");
 
-		ResultSet resultSet = doQueryForResultSet(statement);
+		return doSlice(statement,
+				getRowMapper(entityClass, EntityQueryUtils.getTableName(statement), QueryResultConverter.entity()));
+	}
 
-		Function<Row, T> mapper = getMapper(EntityProjection.nonProjecting(entityClass),
-				EntityQueryUtils.getTableName(statement));
+	<T> Slice<T> doSlice(Statement<?> statement, RowMapper<T> mapper) {
 
-		return EntityQueryUtils.readSlice(resultSet, (row, rowNum) -> mapper.apply(row), 0,
-				getEffectivePageSize(statement));
+		ResultSet resultSet = doQueryForResultSet(statement);
+		return EntityQueryUtils.readSlice(resultSet, mapper, 0, getEffectivePageSize(statement));
 	}
 
 	@Override
@@ -387,9 +396,17 @@ public <T> Stream<T> stream(Statement<?> statement, Class<T> entityClass) throws
 		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 <T> List<T> select(Query query, Class<T> entityClass) throws DataAccessEx
 		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 @@ <T> List<T> doSelect(Query query, Class<?> entityClass, CqlIdentifier tableName,
 		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 <T> Slice<T> slice(Query query, Class<T> entityClass) throws DataAccessEx
 		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());
+	}
+
+	<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 slice(select.build(), entityClass);
+		return doSlice(select.build(), rowMapper);
 	}
 
 	@Override
@@ -445,17 +478,19 @@ public <T> Stream<T> stream(Query query, Class<T> entityClass) throws DataAccess
 		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 <T> ExecutableSelect<T> query(Class<T> domainType) {
 		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 String getCql() {
 		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;
+	}
+}