From 1b0a4da6288062cb3a15dd0b25235090d1750554 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Wed, 16 Apr 2025 16:35:43 +0200 Subject: [PATCH] Add support for QueryResultConverter. Closes #1568 --- .../core/AsyncCassandraOperations.java | 2 +- .../cassandra/core/CassandraOperations.java | 2 +- .../cassandra/core/CassandraTemplate.java | 115 +++++++++-- .../cassandra/core/EntityResultConverter.java | 33 +++ .../core/ExecutableSelectOperation.java | 119 ++++++++++- .../ExecutableSelectOperationSupport.java | 184 ++++++++++++++++- .../cassandra/core/QueryResultConverter.java | 85 ++++++++ .../core/ReactiveCassandraOperations.java | 2 +- .../core/ReactiveCassandraTemplate.java | 46 ++++- .../core/ReactiveSelectOperation.java | 115 ++++++++++- .../core/ReactiveSelectOperationSupport.java | 192 +++++++++++++++--- .../query/CassandraQueryExecution.java | 38 ++++ .../core/CassandraTemplateUnitTests.java | 12 +- ...electOperationSupportIntegrationTests.java | 27 ++- .../core/QueryResultConverterUnitTests.java | 73 +++++++ ...electOperationSupportIntegrationTests.java | 17 +- 16 files changed, 975 insertions(+), 87 deletions(-) create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/EntityResultConverter.java create mode 100644 spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/QueryResultConverter.java create mode 100644 spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/QueryResultConverterUnitTests.java diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraOperations.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraOperations.java index febc49b4f..90cdc8f76 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraOperations.java +++ b/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}. diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java index 943473ee2..4edb52905 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java +++ b/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}. diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java index 9ec3804cb..300bc1c7b 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java +++ b/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 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)); + List doSelect(Statement statement, Class entityClass, CqlIdentifier tableName, Class returnType, + QueryResultConverter mappingFunction) { + + EntityProjection projection = entityOperations.introspectProjection(returnType, entityClass); + + RowMapper 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())); + } + + Slice doSlice(Statement statement, RowMapper mapper) { + ResultSet resultSet = doQueryForResultSet(statement); - - Function 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 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()); + } + + Stream doStream(Statement statement, Class entityClass, CqlIdentifier tableName, Class returnType, + QueryResultConverter mappingFunction) { + + EntityProjection projection = entityOperations.introspectProjection(returnType, entityClass); + + RowMapper 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()); } - List doSelect(Query query, Class entityClass, CqlIdentifier tableName, Class returnType) { + List doSelect(Query query, Class entityClass, CqlIdentifier tableName, Class returnType, + QueryResultConverter mappingFunction) { CassandraPersistentEntity entity = getRequiredPersistentEntity(entityClass); EntityProjection projection = entityOperations.introspectProjection(returnType, entityClass); @@ -415,9 +433,9 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP Query queryToUse = query.columns(columns); StatementBuilder select = getStatementFactory().select(query, getRequiredPersistentEntity(entityClass)); + return doSlice(query, entityClass, getRequiredPersistentEntity(entityClass).getTableName(), entityClass, + QueryResultConverter.entity()); + } - return slice(select.build(), entityClass); + Slice doSlice(Query query, Class entityClass, CqlIdentifier tableName, Class returnType, + QueryResultConverter mappingFunction) { + + CassandraPersistentEntity entity = getRequiredPersistentEntity(entityClass); + EntityProjection projection = entityOperations.introspectProjection(returnType, entityClass); + Columns columns = getStatementFactory().computeColumnsForProjection(projection, query.getColumns(), entity, + returnType); + + Query queryToUse = query.columns(columns); + + StatementBuilder select = getStatementFactory().select(query, getRequiredPersistentEntity(entityClass), tableName); EntityProjection projection = entityOperations.introspectProjection(returnType, entityClass); - Function mapper = getMapper(projection, tableName); - return doQueryForStream(select.build(), (row, rowNum) -> mapper.apply(row)); + RowMapper 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 ExecutableInsert insert(Class domainType) { return new ExecutableInsertOperationSupport(this).insert(domainType); @@ -909,6 +954,32 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP return getCqlOperations().execute(new GetConfiguredPageSize()); } + @SuppressWarnings("unchecked") + RowMapper getRowMapper(EntityProjection projection, CqlIdentifier tableName, + QueryResultConverter mappingFunction) { + + Function mapper = getMapper(projection, tableName); + + return mappingFunction == QueryResultConverter.entity() ? (row, rowNum) -> (R) mapper.apply(row) + : (row, rowNum) -> { + Lazy reader = Lazy.of(() -> mapper.apply(row)); + return mappingFunction.mapRow(row, reader::get); + }; + } + + @SuppressWarnings("unchecked") + RowMapper getRowMapper(Class domainClass, CqlIdentifier tableName, + QueryResultConverter mappingFunction) { + + Function mapper = getMapper(EntityProjection.nonProjecting(domainClass), tableName); + + return mappingFunction == QueryResultConverter.entity() ? (row, rowNum) -> (R) mapper.apply(row) + : (row, rowNum) -> { + Lazy reader = Lazy.of(() -> mapper.apply(row)); + return mappingFunction.mapRow(row, reader::get); + }; + } + @SuppressWarnings("unchecked") private Function getMapper(EntityProjection projection, CqlIdentifier tableName) { diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/EntityResultConverter.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/EntityResultConverter.java new file mode 100644 index 000000000..440c1df75 --- /dev/null +++ b/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 { + + INSTANCE; + + @Override + public Object mapRow(Row row, ConversionResultSupplier reader) { + return reader.get(); + } + + @Override + public QueryResultConverter andThen(QueryResultConverter after) { + return (QueryResultConverter) after; + } +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ExecutableSelectOperation.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ExecutableSelectOperation.java index 54006e963..664159548 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ExecutableSelectOperation.java +++ b/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 { */ ExecutableSelect query(Class 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 {@link Class type} of the result. + * @return new instance of {@link TerminatingResults}. + * @throws IllegalArgumentException if {@link Class resultType} is {@literal null}. + */ + @Contract("_ -> new") + TerminatingResults as(Class 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 {@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 TerminatingResults map(Function 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 {@link Class type} of the result. + * @return new instance of {@link TerminatingResults}. + * @throws IllegalArgumentException if {@link RowMapper mapper} is {@literal null}. + */ + @Contract("_ -> new") + TerminatingResults map(RowMapper mapper); + + } + /** * Table override (optional). */ @@ -121,7 +197,7 @@ public interface ExecutableSelectOperation { * @param {@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 extends TerminatingSelect { /** - * 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 matching(Query query); } @@ -149,7 +226,13 @@ public interface ExecutableSelectOperation { /** * Trigger {@code SELECT} query execution by calling one of the terminating methods. */ - interface TerminatingSelect { + interface TerminatingSelect extends TerminatingProjections, TerminatingResults {} + + /** + * 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 { + + /** + * Map the query result to a different type using {@link QueryResultConverter}. + * + * @param {@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") + TerminatingResults map(QueryResultConverter converter); + /** * Get the first result, or no result. * @@ -214,6 +316,15 @@ public interface ExecutableSelectOperation { */ List 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 slice(); + /** * Stream all matching elements. * diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ExecutableSelectOperationSupport.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ExecutableSelectOperationSupport.java index 2e17e3a67..679c5d2b9 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ExecutableSelectOperationSupport.java +++ b/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 implements ExecutableSelect { + @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 TerminatingResults as(Class resultType) { + + Assert.notNull(resultType, "Result type must not be null"); + + return new TypedSelectSupport<>(template, statement, resultType); + } + + @Override + public TerminatingResults map(RowMapper mapper) { + + Assert.notNull(mapper, "RowMapper must not be null"); + + return new TerminatingSelectResultSupport<>(template, statement, mapper); + } + + } + + static class TypedSelectSupport extends TerminatingSelectResultSupport implements TerminatingResults { + + private final Class domainType; + + TypedSelectSupport(CassandraTemplate template, Statement statement, Class domainType) { + super(template, statement, + template.getRowMapper(domainType, EntityQueryUtils.getTableName(statement), QueryResultConverter.entity())); + + this.domainType = domainType; + } + + @Override + public TerminatingResults map(QueryResultConverter converter) { + + Assert.notNull(converter, "Mapping function must not be null"); + + return new TerminatingSelectResultSupport<>(this.template, this.statement, this.domainType, converter); + } + + } + + static class TerminatingSelectResultSupport implements TerminatingResults { + + final CassandraTemplate template; + + final Statement statement; + + final RowMapper rowMapper; + + TerminatingSelectResultSupport(CassandraTemplate template, Statement statement, RowMapper rowMapper) { + this.template = template; + this.statement = statement; + this.rowMapper = rowMapper; + } + + TerminatingSelectResultSupport(CassandraTemplate template, Statement statement, Class domainType, + QueryResultConverter mappingFunction) { + this(template, statement, + template.getRowMapper(domainType, EntityQueryUtils.getTableName(statement), mappingFunction)); + } + + @Override + public TerminatingResults map(QueryResultConverter 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 result = this.template.getCqlOperations().query(this.statement, this.rowMapper); + + return ObjectUtils.isEmpty(result) ? null : result.iterator().next(); + } + + @Override + public @Nullable T oneValue() { + + List 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 all() { + return this.template.getCqlOperations().query(this.statement, this.rowMapper); + } + + @Override + public Slice slice() { + return this.template.doSlice(this.statement, this.rowMapper); + } + + @Override + public Stream stream() { + return this.template.getCqlOperations().queryForStream(this.statement, this.rowMapper); + } + + } + + static class ExecutableSelectSupport implements ExecutableSelect { private final CassandraTemplate template; private final Class domainType; - private final Class returnType; + private final Class returnType; + + private final QueryResultConverter mappingFunction; private final Query query; private final @Nullable CqlIdentifier tableName; - public ExecutableSelectSupport(CassandraTemplate template, Class domainType, Class returnType, Query query, + public ExecutableSelectSupport(CassandraTemplate template, Class domainType, Class returnType, + QueryResultConverter 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 TerminatingResults map(QueryResultConverter 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 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 result = this.template.doSelect(this.query.limit(1), this.domainType, getTableName(), this.returnType); + List 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 result = this.template.doSelect(this.query.limit(2), this.domainType, getTableName(), this.returnType); + List 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 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 slice() { + return this.template.doSlice(this.query, this.domainType, getTableName(), this.returnType, this.mappingFunction); } @Override public Stream 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 { } + } diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/QueryResultConverter.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/QueryResultConverter.java new file mode 100644 index 000000000..fc08e997b --- /dev/null +++ b/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. + *

+ * 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 object type accepted by this converter. + * @param the returned result type. + * @author Mark Paluch + * @since 5.0 + */ +@FunctionalInterface +public interface QueryResultConverter { + + /** + * Returns a function that returns the materialized entity. + * + * @param the type of the input and output entity to the function. + * @return a function that returns the materialized entity. + */ + @SuppressWarnings("unchecked") + static QueryResultConverter entity() { + return (QueryResultConverter) 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 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 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 QueryResultConverter andThen(QueryResultConverter 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 type of the returned result. + */ + interface ConversionResultSupplier { + + /** + * Obtain the upstream conversion result. + * + * @return the upstream conversion result. + */ + T get(); + + } + +} diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraOperations.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraOperations.java index 311d9dfb6..ef9daa813 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraOperations.java +++ b/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}. diff --git a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java b/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java index 37ae596d0..ae54a522e 100644 --- a/spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java +++ b/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()); } - Flux doSelect(Query query, Class entityClass, CqlIdentifier tableName, Class returnType) { + Flux doSelect(Query query, Class entityClass, CqlIdentifier tableName, Class returnType, + QueryResultConverter mappingFunction) { CassandraPersistentEntity persistentEntity = getRequiredPersistentEntity(entityClass); EntityProjection projection = entityOperations.introspectProjection(returnType, entityClass); @@ -404,9 +406,9 @@ public class ReactiveCassandraTemplate Query queryToUse = query.columns(columns); StatementBuilder