#469 - Suppress emission of null values when using simple and primitive result types.

Results projected onto simple and primitive types that are null are no longer emitted.

A SQL query SELECT MAX(age) FROM my_table that returns a SQL NULL and that would be consumed as Long.class (Publisher<Long>) is an example for a primitive type that can be null. Since Reactive Streams prohibits the propagation of null values by a Publisher to a Subscriber the only viable option is to suppress null results by wrapping the mapping function into Optional result values and filter these values later to avoid null being emitted.

src/main/java/org/springframework/data/r2dbc/convert/MappingR2dbcConverter.java

@@ -579,6 +579,15 @@ public class MappingR2dbcConverter extends BasicRelationalConverter implements R
 		});
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.convert.R2dbcConverter#isSimpleType(Class)
+	 */
+	@Override
+	public boolean isSimpleType(Class<?> type) {
+		return getConversions().isSimpleType(type);
+	}
+
 	// ----------------------------------
 	// Id handling
 	// ----------------------------------

src/main/java/org/springframework/data/r2dbc/convert/R2dbcConverter.java

@@ -72,6 +72,17 @@ public interface R2dbcConverter
 	 */
 	Class<?> getTargetType(Class<?> valueType);
 
+	/**
+	 * Return whether the {@code type} is a simple type. Simple types are database primitives or types with a custom
+	 * mapping strategy.
+	 *
+	 * @param valueType the type to inspect, must not be {@literal null}.
+	 * @return {@literal true} if the type is a simple one.
+	 * @see org.springframework.data.mapping.model.SimpleTypeHolder
+	 * @since 1.2
+	 */
+	boolean isSimpleType(Class<?> type);
+
 	/**
 	 * Returns a {@link java.util.function.Function} that populates the id property of the {@code object} from a
 	 * {@link Row}.

src/main/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplate.java

@@ -432,14 +432,23 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 
 		PreparedOperation<?> operation = statementMapper.getMappedObject(selectSpec);
 
+		boolean simpleType;
 		BiFunction<Row, RowMetadata, T> rowMapper;
 		if (returnType.isInterface()) {
+			simpleType = getConverter().isSimpleType(entityClass);
 			rowMapper = dataAccessStrategy.getRowMapper(entityClass)
 					.andThen(o -> projectionFactory.createProjection(returnType, o));
 		} else {
+			simpleType = getConverter().isSimpleType(returnType);
 			rowMapper = dataAccessStrategy.getRowMapper(returnType);
 		}
 
+		// avoid top-level null values if the read type is a simple one (e.g. SELECT MAX(age) via Integer.class)
+		if (simpleType) {
+			return new UnwrapOptionalFetchSpecAdapter<>(this.databaseClient.sql(operation)
+					.map((row, metadata) -> Optional.ofNullable(rowMapper.apply(row, metadata))));
+		}
+
 		return this.databaseClient.sql(operation).map(rowMapper);
 	}
 
@@ -940,4 +949,28 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 			}
 		}
 	}
+
+	private static class UnwrapOptionalFetchSpecAdapter<T> implements RowsFetchSpec<T> {
+
+		private final RowsFetchSpec<Optional<T>> delegate;
+
+		private UnwrapOptionalFetchSpecAdapter(RowsFetchSpec<Optional<T>> delegate) {
+			this.delegate = delegate;
+		}
+
+		@Override
+		public Mono<T> one() {
+			return delegate.one().handle((optional, sink) -> optional.ifPresent(sink::next));
+		}
+
+		@Override
+		public Mono<T> first() {
+			return delegate.first().handle((optional, sink) -> optional.ifPresent(sink::next));
+		}
+
+		@Override
+		public Flux<T> all() {
+			return delegate.all().handle((optional, sink) -> optional.ifPresent(sink::next));
+		}
+	}
 }

src/main/java/org/springframework/data/r2dbc/repository/query/AbstractR2dbcQuery.java

@@ -18,6 +18,8 @@ package org.springframework.data.r2dbc.repository.query;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.util.Optional;
+
 import org.reactivestreams.Publisher;
 
 import org.springframework.data.mapping.model.EntityInstantiators;
@@ -100,8 +102,17 @@ public abstract class AbstractR2dbcQuery implements RepositoryQuery {
 		if (isExistsQuery()) {
 			fetchSpec = (FetchSpec) boundQuery.map(row -> true);
 		} else if (requiresMapping()) {
-			EntityRowMapper rowMapper = new EntityRowMapper<>(resolveResultType(processor), converter);
-			fetchSpec = new FetchSpecAdapter<>(boundQuery.map(rowMapper));
+
+			Class<?> resultType = resolveResultType(processor);
+			EntityRowMapper rowMapper = new EntityRowMapper<>(resultType, converter);
+
+			if (converter.isSimpleType(resultType)) {
+				fetchSpec = new UnwrapOptionalFetchSpecAdapter<>(
+						boundQuery.map((row, rowMetadata) -> Optional.ofNullable(rowMapper.apply(row, rowMetadata))));
+
+			} else {
+				fetchSpec = new FetchSpecAdapter<>(boundQuery.map(rowMapper));
+			}
 		} else {
 			fetchSpec = (FetchSpec) boundQuery.fetch();
 		}
@@ -222,4 +233,33 @@ public abstract class AbstractR2dbcQuery implements RepositoryQuery {
 			throw new UnsupportedOperationException("Not supported after applying a row mapper");
 		}
 	}
+
+	private static class UnwrapOptionalFetchSpecAdapter<T> implements FetchSpec<T> {
+
+		private final RowsFetchSpec<Optional<T>> delegate;
+
+		private UnwrapOptionalFetchSpecAdapter(RowsFetchSpec<Optional<T>> delegate) {
+			this.delegate = delegate;
+		}
+
+		@Override
+		public Mono<T> one() {
+			return delegate.one().handle((optional, sink) -> optional.ifPresent(sink::next));
+		}
+
+		@Override
+		public Mono<T> first() {
+			return delegate.first().handle((optional, sink) -> optional.ifPresent(sink::next));
+		}
+
+		@Override
+		public Flux<T> all() {
+			return delegate.all().handle((optional, sink) -> optional.ifPresent(sink::next));
+		}
+
+		@Override
+		public Mono<Integer> rowsUpdated() {
+			throw new UnsupportedOperationException("Not supported after applying a row mapper");
+		}
+	}
 }

src/test/java/org/springframework/data/r2dbc/core/R2dbcEntityTemplateUnitTests.java

@@ -104,7 +104,25 @@ public class R2dbcEntityTemplateUnitTests {
 		assertThat(statement.getBindings()).hasSize(1).containsEntry(0, Parameter.from("Walter"));
 	}
 
-	@Test // gh-220
+	@Test // gh-469
+	public void shouldProjectExistsResult() {
+
+		MockRowMetadata metadata = MockRowMetadata.builder()
+				.columnMetadata(MockColumnMetadata.builder().name("name").build()).build();
+		MockResult result = MockResult.builder().rowMetadata(metadata)
+				.row(MockRow.builder().identified(0, Object.class, null).build()).build();
+
+		recorder.addStubbing(s -> s.startsWith("SELECT"), result);
+
+		entityTemplate.select(Person.class) //
+				.as(Integer.class) //
+				.matching(Query.empty().columns("MAX(age)")) //
+				.all() //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+	}
+
+	@Test // gh-469
 	public void shouldExistsByCriteria() {
 
 		MockRowMetadata metadata = MockRowMetadata.builder()

src/test/java/org/springframework/data/r2dbc/repository/H2R2dbcRepositoryIntegrationTests.java

@@ -92,6 +92,11 @@ public class H2R2dbcRepositoryIntegrationTests extends AbstractR2dbcRepositoryIn
 		return H2LegoSetRepository.class;
 	}
 
+	@Test // gh-469
+	public void shouldSuppressNullValues() {
+		repository.findMax("doo").as(StepVerifier::create).verifyComplete();
+	}
+
 	@Test // gh-235
 	public void shouldReturnUpdateCount() {
 
@@ -139,6 +144,9 @@ public class H2R2dbcRepositoryIntegrationTests extends AbstractR2dbcRepositoryIn
 
 	interface H2LegoSetRepository extends LegoSetRepository {
 
+		@Query("SELECT MAX(manual) FROM legoset WHERE name = :name")
+		Mono<Integer> findMax(String name);
+
 		@Override
 		@Query("SELECT name FROM legoset")
 		Flux<Named> findAsProjection();