Fix Composite ids for R2DBC.

R2DBC now has minimal support for embedded entities.
We can read and write them.
And we can use them as ids.

Closes #2012
Original pull request: #2114

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

@@ -57,6 +57,7 @@ import org.springframework.util.CollectionUtils;
  *
  * @author Mark Paluch
  * @author Oliver Drotbohm
+ * @author Jens Schauder
  */
 public class MappingR2dbcConverter extends MappingRelationalConverter implements R2dbcConverter {
 
@@ -189,8 +190,17 @@ public class MappingR2dbcConverter extends MappingRelationalConverter implements
 		writeProperties(sink, entity, propertyAccessor);
 	}
 
+	/**
+	 * write the values of the properties of an {@link RelationalPersistentEntity} to an {@link OutboundRow}.
+	 * 
+	 * @param sink must not be {@literal null}.
+	 * @param entity must not be {@literal null}.
+	 * @param accessor used for accessing the property values of {@literal entity}. May be {@literal null}. A
+	 *          {@literal null} value is used when this is an embedded {@literal null} entity, resulting in all its
+	 *          property values to be {@literal null} as well.
+	 */
 	private void writeProperties(OutboundRow sink, RelationalPersistentEntity<?> entity,
-			PersistentPropertyAccessor<?> accessor) {
+			@Nullable PersistentPropertyAccessor<?> accessor) {
 
 		for (RelationalPersistentProperty property : entity) {
 
@@ -200,11 +210,27 @@ public class MappingR2dbcConverter extends MappingRelationalConverter implements
 
 			Object value;
 
-			if (property.isIdProperty()) {
-				IdentifierAccessor identifierAccessor = entity.getIdentifierAccessor(accessor.getBean());
-				value = identifierAccessor.getIdentifier();
+			if (accessor == null) {
+				value = null;
 			} else {
-				value = accessor.getProperty(property);
+				if (property.isIdProperty()) {
+					IdentifierAccessor identifierAccessor = entity.getIdentifierAccessor(accessor.getBean());
+					value = identifierAccessor.getIdentifier();
+				} else {
+					value = accessor.getProperty(property);
+				}
+			}
+
+			if (property.isEmbedded()) {
+
+				RelationalPersistentEntity<?> embeddedEntity = getMappingContext().getRequiredPersistentEntity(property);
+				PersistentPropertyAccessor<Object> embeddedAccessor = null;
+				if (value != null) {
+					embeddedAccessor = embeddedEntity.getPropertyAccessor(value);
+				}
+				writeProperties(sink, embeddedEntity, embeddedAccessor);
+
+				continue;
 			}
 
 			if (value == null) {

spring-data-r2dbc/src/main/java/org/springframework/data/r2dbc/core/DefaultReactiveDataAccessStrategy.java

@@ -43,6 +43,7 @@ import org.springframework.data.r2dbc.support.ArrayUtils;
 import org.springframework.data.relational.core.dialect.ArrayColumns;
 import org.springframework.data.relational.core.dialect.Dialect;
 import org.springframework.data.relational.core.dialect.RenderContextFactory;
+import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
@@ -66,7 +67,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 	private final R2dbcDialect dialect;
 	private final R2dbcConverter converter;
 	private final UpdateMapper updateMapper;
-	private final MappingContext<RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty> mappingContext;
+	private final RelationalMappingContext mappingContext;
 	private final StatementMapper statementMapper;
 	private final NamedParameterExpander expander = new NamedParameterExpander();
 
@@ -119,7 +120,6 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 	 * @param dialect the {@link R2dbcDialect} to use.
 	 * @param converter must not be {@literal null}.
 	 */
-	@SuppressWarnings("unchecked")
 	public DefaultReactiveDataAccessStrategy(R2dbcDialect dialect, R2dbcConverter converter) {
 
 		Assert.notNull(dialect, "Dialect must not be null");
@@ -127,8 +127,7 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 
 		this.converter = converter;
 		this.updateMapper = new UpdateMapper(dialect, converter);
-		this.mappingContext = (MappingContext<RelationalPersistentEntity<?>, ? extends RelationalPersistentProperty>) this.converter
-				.getMappingContext();
+		this.mappingContext = (RelationalMappingContext) this.converter.getMappingContext();
 		this.dialect = dialect;
 
 		RenderContextFactory factory = new RenderContextFactory(dialect);
@@ -141,13 +140,22 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 
 		RelationalPersistentEntity<?> persistentEntity = getPersistentEntity(entityType);
 
+		return getAllColumns(persistentEntity);
+	}
+
+	private List<SqlIdentifier> getAllColumns(@Nullable RelationalPersistentEntity<?> persistentEntity) {
+
 		if (persistentEntity == null) {
 			return Collections.singletonList(SqlIdentifier.unquoted("*"));
 		}
 
 		List<SqlIdentifier> columnNames = new ArrayList<>();
 		for (RelationalPersistentProperty property : persistentEntity) {
-			columnNames.add(property.getColumnName());
+			if (property.isEmbedded()) {
+				columnNames.addAll(getAllColumns(mappingContext.getRequiredPersistentEntity(property)));
+			} else {
+				columnNames.add(property.getColumnName());
+			}
 		}
 
 		return columnNames;
@@ -159,12 +167,8 @@ public class DefaultReactiveDataAccessStrategy implements ReactiveDataAccessStra
 		RelationalPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entityType);
 
 		List<SqlIdentifier> columnNames = new ArrayList<>();
-		for (RelationalPersistentProperty property : persistentEntity) {
-
-			if (property.isIdProperty()) {
-				columnNames.add(property.getColumnName());
-			}
-		}
+		mappingContext.getAggregatePath(persistentEntity).getTableInfo().idColumnInfos()
+				.forEach((__, ci) -> columnNames.add(ci.name()));
 
 		return columnNames;
 	}

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

@@ -23,6 +23,7 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
@@ -33,7 +34,6 @@ import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import org.reactivestreams.Publisher;
-
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
@@ -60,6 +60,7 @@ import org.springframework.data.r2dbc.mapping.event.BeforeConvertCallback;
 import org.springframework.data.r2dbc.mapping.event.BeforeSaveCallback;
 import org.springframework.data.relational.core.conversion.AbstractRelationalConverter;
 import org.springframework.data.relational.core.mapping.PersistentPropertyTranslator;
+import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.relational.core.query.Criteria;
@@ -96,6 +97,7 @@ import org.springframework.util.Assert;
  * @author Robert Heim
  * @author Sebastian Wieland
  * @author Mikhail Polivakha
+ * @author Jens Schauder
  * @since 1.1
  */
 public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAware, ApplicationContextAware {
@@ -350,8 +352,8 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 		return (P) ((Flux<?>) result).concatMap(it -> maybeCallAfterConvert(it, tableName));
 	}
 
-	private <T> RowsFetchSpec<T> doSelect(Query query, Class<?> entityType, SqlIdentifier tableName,
-			Class<T> returnType, Function<? super Statement, ? extends Statement> filterFunction) {
+	private <T> RowsFetchSpec<T> doSelect(Query query, Class<?> entityType, SqlIdentifier tableName, Class<T> returnType,
+			Function<? super Statement, ? extends Statement> filterFunction) {
 
 		StatementMapper statementMapper = dataAccessStrategy.getStatementMapper().forType(entityType);
 
@@ -378,11 +380,8 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 
 		PreparedOperation<?> operation = statementMapper.getMappedObject(selectSpec);
 
-		return getRowsFetchSpec(
-				databaseClient.sql(operation).filter(statementFilterFunction.andThen(filterFunction)),
-			entityType,
-			returnType
-		);
+		return getRowsFetchSpec(databaseClient.sql(operation).filter(statementFilterFunction.andThen(filterFunction)),
+				entityType, returnType);
 	}
 
 	@Override
@@ -622,8 +621,9 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 			return maybeCallBeforeSave(entityToUse, outboundRow, tableName) //
 					.flatMap(onBeforeSave -> {
 
-						SqlIdentifier idColumn = persistentEntity.getRequiredIdProperty().getColumnName();
-						Parameter id = outboundRow.remove(idColumn);
+						Map<SqlIdentifier, Object> idValues = new HashMap<>();
+						((RelationalMappingContext) mappingContext).getAggregatePath(persistentEntity).getTableInfo()
+								.idColumnInfos().forEach((ap, ci) -> idValues.put(ci.name(), outboundRow.remove(ci.name())));
 
 						persistentEntity.forEach(p -> {
 							if (p.isInsertOnly()) {
@@ -631,7 +631,16 @@ public class R2dbcEntityTemplate implements R2dbcEntityOperations, BeanFactoryAw
 							}
 						});
 
-						Criteria criteria = Criteria.where(dataAccessStrategy.toSql(idColumn)).is(id);
+						Assert.state(!idValues.isEmpty(), entityToUse + " has no id. Update is not possible");
+
+						Criteria criteria = null;
+						for (Map.Entry<SqlIdentifier, Object> idAndValue : idValues.entrySet()) {
+							if (criteria == null) {
+								criteria = Criteria.where(dataAccessStrategy.toSql(idAndValue.getKey())).is(idAndValue.getValue());
+							} else {
+								criteria = criteria.and(dataAccessStrategy.toSql(idAndValue.getKey())).is(idAndValue.getValue());
+							}
+						}
 
 						if (matchingVersionCriteria != null) {
 							criteria = criteria.and(matchingVersionCriteria);

spring-data-r2dbc/src/test/java/org/springframework/data/r2dbc/convert/MappingR2dbcConverterUnitTests.java

@@ -43,6 +43,7 @@ import org.springframework.data.domain.Persistable;
 import org.springframework.data.r2dbc.dialect.PostgresDialect;
 import org.springframework.data.r2dbc.mapping.OutboundRow;
 import org.springframework.data.r2dbc.mapping.R2dbcMappingContext;
+import org.springframework.data.relational.core.mapping.Embedded;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.r2dbc.core.Parameter;
@@ -261,6 +262,53 @@ public class MappingR2dbcConverterUnitTests {
 		assertThat(row).containsEntry(SqlIdentifier.unquoted("id"), Parameter.from(42L));
 	}
 
+	@Test // GH-2096
+	void shouldWriteSingleLevelEmbeddedEntity() {
+
+		Level1 entity = new Level1("root", new Level2("child", 23));
+
+		OutboundRow row = new OutboundRow();
+		converter.write(entity, row);
+
+		assertThat(row).containsExactlyInAnyOrderEntriesOf(Map.of(
+				SqlIdentifier.unquoted("name"), Parameter.from("root"),
+				SqlIdentifier.unquoted("level2_name"), Parameter.from("child"),
+				SqlIdentifier.unquoted("level2_number"), Parameter.from(23)
+		));
+	}
+
+	@Test // GH-2096
+	void shouldWriteMultiLevelEmbeddedEntity() {
+
+		WithEmbedded entity = new WithEmbedded(4711L, new Level1("level1", new Level2("child", 23)));
+
+		OutboundRow row = new OutboundRow();
+		converter.write(entity, row);
+
+		assertThat(row).containsExactlyInAnyOrderEntriesOf(Map.of(
+				SqlIdentifier.unquoted("id"), Parameter.from(4711L),
+				SqlIdentifier.unquoted("level1_name"), Parameter.from("level1"),
+				SqlIdentifier.unquoted("level1_level2_name"), Parameter.from("child"),
+				SqlIdentifier.unquoted("level1_level2_number"), Parameter.from(23)
+		));
+	}
+
+	@Test // GH-2096
+	void shouldWriteNullEmbeddedEntity() {
+
+		WithEmbedded entity = new WithEmbedded(4711L, null);
+
+		OutboundRow row = new OutboundRow();
+		converter.write(entity, row);
+
+		assertThat(row).containsExactlyInAnyOrderEntriesOf(Map.of(
+				SqlIdentifier.unquoted("id"), Parameter.from(4711L),
+				SqlIdentifier.unquoted("level1_name"), Parameter.empty(String.class),
+				SqlIdentifier.unquoted("level1_level2_name"), Parameter.empty(String.class),
+				SqlIdentifier.unquoted("level1_level2_number"), Parameter.empty(Integer.class)
+		));
+	}
+
 	static class Person {
 		@Id String id;
 		String firstname, lastname;
@@ -326,6 +374,13 @@ public class MappingR2dbcConverterUnitTests {
 	record WithPrimitiveId(@Id long id) {
 	}
 
+	record WithEmbedded(@Id long id, @Embedded.Empty(prefix = "level1_") Level1 one){}
+
+	record Level1(String name, @Embedded.Empty(prefix = "level2_") Level2 two) {
+
+	}
+	record Level2(String name, Integer number){}
+
 	static class CustomConversionPerson {
 
 		String foo;

spring-data-r2dbc/src/test/java/org/springframework/data/r2dbc/core/DefaultReactiveDataAccessStrategyUnitTests.java

@@ -0,0 +1,54 @@
+package org.springframework.data.r2dbc.core;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.assertj.core.api.SoftAssertions;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.r2dbc.dialect.H2Dialect;
+import org.springframework.data.relational.core.mapping.Embedded;
+import org.springframework.data.relational.core.sql.SqlIdentifier;
+
+/**
+ * Unit tests for {@link DefaultReactiveDataAccessStrategy}.
+ *
+ * @author Jens Schauder
+ */
+class DefaultReactiveDataAccessStrategyUnitTests {
+
+	DefaultReactiveDataAccessStrategy dataAccessStrategy = new DefaultReactiveDataAccessStrategy(H2Dialect.INSTANCE);
+
+	@Test
+	void getAllColumns() {
+
+		SoftAssertions.assertSoftly(softly -> {
+			check(softly, SimpleEntity.class, "ID", "NAME");
+			check(softly, WithEmbedded.class, "ID", "L1_NAME", "L1_L2_NAME", "L1_L2_NUMBER");
+			check(softly, WithEmbeddedId.class, "ID_NAME", "ID_NUMBER", "NAME");
+		});
+	}
+
+	private void check(SoftAssertions softly, Class<?> entityType, String... columnNames) {
+
+		List<SqlIdentifier> sqlIdentifiers = Arrays.stream(columnNames).map(SqlIdentifier::quoted).toList();
+		softly.assertThat(dataAccessStrategy.getAllColumns(entityType)).describedAs(entityType.getName())
+				.containsExactlyInAnyOrder(sqlIdentifiers.toArray(new SqlIdentifier[0]));
+	}
+
+	record SimpleEntity(int id, String name) {
+	}
+
+	record WithEmbedded(int id, @Embedded.Empty(prefix = "L1_") Level1 level1) {
+	}
+
+	record Level1(String name, @Embedded.Empty(prefix = "L2_") Level2 l2) {
+	}
+
+	record Level2(String name, Integer number) {
+	}
+
+	record WithEmbeddedId(@Id @Embedded.Empty(prefix = "ID_") Level2 id, String name) {
+	}
+
+}

spring-data-r2dbc/src/test/java/org/springframework/data/r2dbc/repository/CompositeIdRepositoryIntegrationTests.java

@@ -18,16 +18,21 @@ package org.springframework.data.r2dbc.repository;
 import static org.assertj.core.api.Assertions.*;
 
 import io.r2dbc.spi.ConnectionFactory;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import javax.sql.DataSource;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.reactivestreams.Publisher;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.FilterType;
@@ -36,12 +41,14 @@ import org.springframework.data.annotation.Id;
 import org.springframework.data.r2dbc.config.AbstractR2dbcConfiguration;
 import org.springframework.data.r2dbc.convert.R2dbcCustomConversions;
 import org.springframework.data.r2dbc.mapping.R2dbcMappingContext;
+import org.springframework.data.r2dbc.mapping.event.BeforeConvertCallback;
 import org.springframework.data.r2dbc.repository.config.EnableR2dbcRepositories;
 import org.springframework.data.r2dbc.testing.H2TestSupport;
 import org.springframework.data.relational.RelationalManagedTypes;
 import org.springframework.data.relational.core.mapping.Embedded;
 import org.springframework.data.relational.core.mapping.NamingStrategy;
 import org.springframework.data.relational.core.mapping.Table;
+import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.data.repository.reactive.ReactiveCrudRepository;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
@@ -76,6 +83,24 @@ public class CompositeIdRepositoryIntegrationTests {
 
 			return context;
 		}
+
+		@Bean
+		BeforeConvertCallback<WithCompositeId> beforeConvertCallback() {
+
+			return new BeforeConvertCallback<>() {
+				AtomicInteger counter = new AtomicInteger();
+
+				@Override
+				public Publisher<WithCompositeId> onBeforeConvert(WithCompositeId entity, SqlIdentifier table) {
+
+					if (entity.pk == null) {
+						CompositeId pk = new CompositeId(counter.incrementAndGet(), "generated");
+						entity = new WithCompositeId(pk, entity.name);
+					}
+					return Mono.just(entity);
+				}
+			};
+		}
 	}
 
 	@BeforeEach
@@ -117,15 +142,71 @@ public class CompositeIdRepositoryIntegrationTests {
 
 	@Test // GH-574
 	void findAllById() {
+
 		repository.findById(new CompositeId(42, "HBAR")) //
 				.as(StepVerifier::create) //
-				.consumeNextWith(actual -> {
+				.assertNext(actual -> {
 					assertThat(actual.name).isEqualTo("Walter");
+					assertThat(actual.pk.one).isEqualTo(42);
+					assertThat(actual.pk.two).isEqualTo("HBAR");
 				}).verifyComplete();
 	}
 
-	interface WithCompositeIdRepository extends ReactiveCrudRepository<WithCompositeId, CompositeId> {
+	@Test // GH-2096
+	void findByName() {
 
+		repository.findByName("Walter") //
+				.as(StepVerifier::create) //
+				.assertNext(actual -> {
+					assertThat(actual.name).isEqualTo("Walter");
+					assertThat(actual.pk.one).isEqualTo(42);
+					assertThat(actual.pk.two).isEqualTo("HBAR");
+				}).verifyComplete();
+	}
+
+	@Test // GH-2096
+	void insert() {
+
+		repository.save(new WithCompositeId(null, "Jane Margolis"))//
+				.as(StepVerifier::create) //
+				.assertNext(actual -> assertThat(actual.pk).isNotNull()).verifyComplete();
+	}
+
+	@Test // GH-2096
+	void update() {
+
+		insert();
+
+		repository.findByName("Jane Margolis") //
+				.map(wci -> new WithCompositeId(wci.pk, "Jane")) //
+				.flatMap(repository::save) //
+				.as(StepVerifier::create) //
+				.expectNextCount(1) //
+				.verifyComplete();
+
+		// nothing to be found under the old name
+		repository.findByName("Jane Margolis").as(StepVerifier::create).verifyComplete();
+
+		// but under the new name
+		repository.findByName("Jane").as(StepVerifier::create).expectNextCount(1).verifyComplete();
+	}
+
+	@Test
+	void delete() {
+
+		insert();
+
+		repository.findByName("Jane Margolis") //
+				.flatMap(repository::delete) //
+				.as(StepVerifier::create) //
+				.verifyComplete();
+
+		// nothing to be found under the old name
+		repository.findByName("Jane Margolis").as(StepVerifier::create).verifyComplete();
+	}
+
+	interface WithCompositeIdRepository extends ReactiveCrudRepository<WithCompositeId, CompositeId> {
+		Flux<WithCompositeId> findByName(String name);
 	}
 
 	@Table("with_composite_id")

spring-data-r2dbc/src/test/java/org/springframework/data/r2dbc/repository/H2R2dbcRepositoryEmbeddedIntegrationTests.java

@@ -0,0 +1,159 @@
+/*
+ * Copyright 2018-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.r2dbc.repository;
+
+import io.r2dbc.spi.ConnectionFactory;
+import reactor.core.publisher.Hooks;
+import reactor.test.StepVerifier;
+
+import java.util.Arrays;
+import java.util.Optional;
+import java.util.Set;
+
+import javax.sql.DataSource;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.FilterType;
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.r2dbc.config.AbstractR2dbcConfiguration;
+import org.springframework.data.r2dbc.convert.R2dbcCustomConversions;
+import org.springframework.data.r2dbc.mapping.R2dbcMappingContext;
+import org.springframework.data.r2dbc.repository.config.EnableR2dbcRepositories;
+import org.springframework.data.r2dbc.testing.H2TestSupport;
+import org.springframework.data.r2dbc.testing.R2dbcIntegrationTestSupport;
+import org.springframework.data.relational.RelationalManagedTypes;
+import org.springframework.data.relational.core.mapping.Embedded;
+import org.springframework.data.relational.core.mapping.NamingStrategy;
+import org.springframework.data.repository.reactive.ReactiveCrudRepository;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+
+/**
+ * Tests for support of embedded entities.
+ *
+ * @author Jens Schauder
+ */
+@ExtendWith(SpringExtension.class)
+@ContextConfiguration
+class H2R2dbcRepositoryEmbeddedIntegrationTests extends R2dbcIntegrationTestSupport {
+
+	static {
+		Hooks.onOperatorDebug();
+	}
+
+	@Autowired private PersonRepository repository;
+	@Autowired private ConnectionFactory connectionFactory;
+	protected JdbcTemplate jdbc;
+
+	@Configuration
+	@EnableR2dbcRepositories(considerNestedRepositories = true,
+			includeFilters = @ComponentScan.Filter(classes = PersonRepository.class, type = FilterType.ASSIGNABLE_TYPE))
+	static class IntegrationTestConfiguration extends AbstractR2dbcConfiguration {
+
+		@Bean
+		@Override
+		public ConnectionFactory connectionFactory() {
+			return H2TestSupport.createConnectionFactory();
+		}
+
+		@Override
+		public R2dbcMappingContext r2dbcMappingContext(Optional<NamingStrategy> namingStrategy,
+				R2dbcCustomConversions r2dbcCustomConversions, RelationalManagedTypes r2dbcManagedTypes) {
+
+			R2dbcMappingContext context = super.r2dbcMappingContext(namingStrategy, r2dbcCustomConversions,
+					r2dbcManagedTypes);
+			context.setForceQuote(false);
+
+			return context;
+		}
+
+		@Bean
+		public H2R2dbcRepositoryIntegrationTests.AfterConvertCallbackRecorder afterConvertCallbackRecorder() {
+			return new H2R2dbcRepositoryIntegrationTests.AfterConvertCallbackRecorder();
+		}
+	}
+
+	@BeforeEach
+	void before() {
+
+		this.jdbc = createJdbcTemplate(createDataSource());
+
+		try {
+			this.jdbc.execute("DROP TABLE person");
+		} catch (DataAccessException e) {}
+
+		this.jdbc.execute(getCreateTableStatement());
+	}
+
+	/**
+	 * Creates a {@link DataSource} to be used in this test.
+	 *
+	 * @return the {@link DataSource} to be used in this test.
+	 */
+	DataSource createDataSource() {
+		return H2TestSupport.createDataSource();
+	}
+
+	String getCreateTableStatement() {
+		return "create table person(id integer AUTO_INCREMENT PRIMARY KEY, name_first varchar(50), name_last varchar(50))";
+	}
+
+	@Test // GH-2096
+	void shouldInsertNewItems() {
+
+		Person frodo = new Person(null, new Name("Frodo", "Baggins"));
+		Person sam = new Person(null, new Name("Sam", "Gamgee"));
+
+		repository.saveAll(Arrays.asList(frodo, sam)) //
+				.as(StepVerifier::create) //
+				.expectNextMatches(person -> person.id != null) //
+				.expectNextMatches(person -> person.id != null) //
+				.verifyComplete();
+	}
+
+	@Test // GH-2096
+	void shouldReadNewItems() {
+
+		shouldInsertNewItems();
+
+		Set<String> firstNames = Set.of("Frodo", "Sam");
+
+		repository.findAll() //
+				.as(StepVerifier::create) //
+				.assertNext(p -> firstNames.contains(p.name.first)) //
+				.assertNext(p -> firstNames.contains(p.name.first)) //
+				.verifyComplete();
+	}
+
+	interface PersonRepository extends ReactiveCrudRepository<Person, Integer> {}
+
+	record Person(@Id Integer id, @Embedded.Empty(prefix = "name_") Name name) {
+
+	}
+
+	record Name(String first, String last) {
+
+	}
+
+}