Add initial support for upsert.

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-relational-parent</artifactId>
-	<version>4.1.0-SNAPSHOT</version>
+	<version>4.1.x-GH-493-SNAPSHOT</version>
 	<packaging>pom</packaging>
 
 	<name>Spring Data Relational Parent</name>

spring-data-jdbc-distribution/pom.xml

@@ -14,7 +14,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-relational-parent</artifactId>
-		<version>4.1.0-SNAPSHOT</version>
+		<version>4.1.x-GH-493-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-jdbc/pom.xml

@@ -6,7 +6,7 @@
 	<modelVersion>4.0.0</modelVersion>
 
 	<artifactId>spring-data-jdbc</artifactId>
-	<version>4.1.0-SNAPSHOT</version>
+	<version>4.1.x-GH-493-SNAPSHOT</version>
 
 	<name>Spring Data JDBC</name>
 	<description>Spring Data module for JDBC repositories.</description>
@@ -15,7 +15,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-relational-parent</artifactId>
-		<version>4.1.0-SNAPSHOT</version>
+		<version>4.1.x-GH-493-SNAPSHOT</version>
 	</parent>
 
 	<properties>

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/AggregateChangeExecutor.java

@@ -30,6 +30,7 @@
  * @author Myeonghyeon Lee
  * @author Chirag Tailor
  * @author Mikhail Polivakha
+ * @author Christoph Strobl
  * @since 2.0
  */
 class AggregateChangeExecutor {
@@ -87,6 +88,8 @@ private void execute(DbAction<?> action, JdbcAggregateChangeExecutionContext exe
 			executionContext.executeInsert(insert);
 		} else if (action instanceof DbAction.BatchInsert<?> batchInsert) {
 			executionContext.executeBatchInsert(batchInsert);
+		} else if (action instanceof DbAction.UpsertRoot<?> upsertRoot) {
+			executionContext.executeUpsertRoot(upsertRoot);
 		} else if (action instanceof DbAction.UpdateRoot<?> updateRoot) {
 			executionContext.executeUpdateRoot(updateRoot);
 		} else if (action instanceof DbAction.Delete<?> delete) {

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateChangeExecutionContext.java

@@ -60,6 +60,7 @@
  * @author Myeonghyeon Lee
  * @author Chirag Tailor
  * @author Mark Paluch
+ * @author Christoph Strobl
  */
 @SuppressWarnings("rawtypes")
 class JdbcAggregateChangeExecutionContext {
@@ -120,6 +121,17 @@ <T> void executeBatchInsert(DbAction.BatchInsert<T> batchInsert) {
 		}
 	}
 
+	/**
+	 * @param upsert
+	 * @param <T>
+	 * @since 4.x
+	 */
+	<T> void executeUpsertRoot(DbAction.UpsertRoot<T> upsert) {
+
+		accessStrategy.upsert(upsert.entity(), upsert.getEntityType());
+		add(new DbActionExecutionResult(upsert));
+	}
+
 	<T> void executeUpdateRoot(DbAction.UpdateRoot<T> update) {
 
 		if (update.getPreviousVersion() != null) {
@@ -276,7 +288,8 @@ <T> List<T> populateIdsIfNecessary() {
 
 			Object newEntity = setIdAndCascadingProperties(action, result.getGeneratedId(), cascadingValues);
 
-			if (action instanceof DbAction.InsertRoot || action instanceof DbAction.UpdateRoot) {
+			if (action instanceof DbAction.InsertRoot || action instanceof DbAction.UpdateRoot
+					|| action instanceof DbAction.UpsertRoot) {
 				// noinspection unchecked
 				roots.add((T) newEntity);
 			}
@@ -299,8 +312,9 @@ <T> List<T> populateIdsIfNecessary() {
 
 		if (roots.isEmpty()) {
 			throw new IllegalStateException(
-					String.format("Cannot retrieve the resulting instance(s) unless a %s or %s action was successfully executed",
-							DbAction.InsertRoot.class.getName(), DbAction.UpdateRoot.class.getName()));
+					String.format("Cannot retrieve the resulting instance(s) unless a %s, %s, or %s action was successfully executed",
+							DbAction.InsertRoot.class.getName(), DbAction.UpdateRoot.class.getName(),
+							DbAction.UpsertRoot.class.getName()));
 		}
 
 		Collections.reverse(roots);
@@ -345,6 +359,10 @@ private PersistentPropertyPath<?> getRelativePath(DbAction<?> action, Persistent
 			return pathToValue;
 		}
 
+		if (action instanceof DbAction.UpsertRoot) {
+			return pathToValue;
+		}
+
 		throw new IllegalArgumentException(String.format("DbAction of type %s is not supported", action.getClass()));
 	}
 

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateOperations.java

@@ -115,6 +115,18 @@ public interface JdbcAggregateOperations {
 	 */
 	<T> List<T> updateAll(Iterable<T> instances);
 
+	/**
+	 * Upserts a single aggregate root (insert if row for id does not exist, update if it exists). The instance must have
+	 * an id set. Only supported when the dialect supports single-statement upsert.
+	 *
+	 * @param instance the aggregate root to upsert. Must not be {@code null}. Must have an id set.
+	 * @param <T> the type of the aggregate root.
+	 * @return the same instance (possibly with generated id set if the dialect returns one).
+	 * @throws UnsupportedOperationException if the dialect does not support upsert.
+	 * @since 4.x
+	 */
+	<T> T upsert(T instance);
+
 	/**
 	 * Counts the number of aggregates of a given type.
 	 *

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/JdbcAggregateTemplate.java

@@ -41,6 +41,7 @@
 import org.springframework.data.domain.Sort;
 import org.springframework.data.jdbc.core.convert.DataAccessStrategy;
 import org.springframework.data.jdbc.core.convert.EntityRowMapper;
+import org.springframework.data.jdbc.core.convert.Identifier;
 import org.springframework.data.jdbc.core.convert.JdbcConverter;
 import org.springframework.data.mapping.IdentifierAccessor;
 import org.springframework.data.mapping.callback.EntityCallbacks;
@@ -52,6 +53,7 @@
 import org.springframework.data.relational.core.conversion.RelationalEntityDeleteWriter;
 import org.springframework.data.relational.core.conversion.RelationalEntityInsertWriter;
 import org.springframework.data.relational.core.conversion.RelationalEntityUpdateWriter;
+import org.springframework.data.relational.core.conversion.RelationalEntityUpsertWriter;
 import org.springframework.data.relational.core.conversion.RelationalEntityVersionUtils;
 import org.springframework.data.relational.core.conversion.RootAggregateChange;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
@@ -266,6 +268,14 @@ public <T> List<T> updateAll(Iterable<T> instances) {
 		return doInBatch(instances, entity -> createUpdateChange(prepareVersionForUpdate(entity)));
 	}
 
+	@Override
+	public <T> T upsert(T instance) {
+
+		Assert.notNull(instance, "Aggregate instance must not be null");
+
+		return performSave(new EntityAndChangeCreator<>(instance, entity -> createUpsertChange(entity)));
+	}
+
 	private <T> List<T> saveInBatch(Iterable<T> instances, Function<T, AggregateChangeCreator<T>> changes) {
 
 		Assert.notNull(instances, "Aggregate instances must not be null");
@@ -622,6 +632,13 @@ private <T> AggregateChangeCreator<T> changeCreatorSelectorForSave(T instance) {
 				: entity -> createUpdateChange(prepareVersionForUpdate(entity));
 	}
 
+	private <T> RootAggregateChange<T> createUpsertChange(T instance) {
+
+		RootAggregateChange<T> aggregateChange = MutableAggregateChange.forSave(instance);
+		new RelationalEntityUpsertWriter<T>(context).write(instance, aggregateChange);
+		return aggregateChange;
+	}
+
 	private <T> RootAggregateChange<T> createInsertChange(T instance) {
 
 		RootAggregateChange<T> aggregateChange = MutableAggregateChange.forSave(instance);
@@ -734,7 +751,8 @@ private <T> T triggerAfterSave(T aggregateRoot, AggregateChange<T> change) {
 
 	private <T> void triggerAfterDelete(@Nullable T aggregateRoot, Object id, AggregateChange<T> change) {
 
-		eventDelegate.publishEvent(() -> new AfterDeleteEvent<>(Identifier.of(id), aggregateRoot, change));
+		eventDelegate.publishEvent(() -> new AfterDeleteEvent<>(
+				org.springframework.data.relational.core.mapping.event.Identifier.of(id), aggregateRoot, change));
 
 		if (aggregateRoot != null && entityCallbacks != null) {
 			entityCallbacks.callback(AfterDeleteCallback.class, aggregateRoot);
@@ -744,7 +762,8 @@ private <T> void triggerAfterDelete(@Nullable T aggregateRoot, Object id, Aggreg
 	@Nullable
 	private <T> T triggerBeforeDelete(@Nullable T aggregateRoot, Object id, MutableAggregateChange<T> change) {
 
-		eventDelegate.publishEvent(() -> new BeforeDeleteEvent<>(Identifier.of(id), aggregateRoot, change));
+		eventDelegate.publishEvent(() -> new BeforeDeleteEvent<>(
+				org.springframework.data.relational.core.mapping.event.Identifier.of(id), aggregateRoot, change));
 
 		if (aggregateRoot != null && entityCallbacks != null) {
 			return entityCallbacks.callback(BeforeDeleteCallback.class, aggregateRoot, change);

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/Association.java

@@ -91,6 +91,11 @@ public static Association from(RelationalPersistentProperty property, JdbcConver
 	}
 
 	private static boolean hasMultipleColumns(@Nullable RelationalPersistentEntity<?> identifierEntity) {
+
+		if( identifierEntity == null ) {
+			return false;
+		}
+
 		Iterator<RelationalPersistentProperty> iterator = identifierEntity.iterator();
 		if (iterator.hasNext()) {
 			iterator.next();

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/CascadingDataAccessStrategy.java

@@ -15,7 +15,7 @@
  */
 package org.springframework.data.jdbc.core.convert;
 
-import static java.lang.Boolean.*;
+import static java.lang.Boolean.TRUE;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -25,7 +25,6 @@
 import java.util.stream.Stream;
 
 import org.jspecify.annotations.Nullable;
-
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.mapping.PersistentPropertyPath;
@@ -49,6 +48,7 @@
  * @author Chirag Tailor
  * @author Diego Krupitza
  * @author Sergey Korotaev
+ * @author Christoph Strobl
  * @since 1.1
  */
 public class CascadingDataAccessStrategy implements DataAccessStrategy {
@@ -87,6 +87,11 @@ public NamedParameterJdbcOperations getJdbcOperations() {
 		return collect(das -> das.insert(insertSubjects, domainType, idValueSource));
 	}
 
+	@Override
+	public <T> int upsert(T instance, Class<? super T> domainType) {
+		return collect(das -> das.upsert(instance, domainType));
+	}
+
 	@Override
 	public <S> boolean update(S instance, Class<S> domainType) {
 		return collect(das -> das.update(instance, domainType));

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DataAccessStrategy.java

@@ -45,6 +45,7 @@
  * @author Chirag Tailor
  * @author Diego Krupitza
  * @author Sergey Korotaev
+ * @author Christoph Strobl
  */
 public interface DataAccessStrategy extends ReadingDataAccessStrategy, RelationResolver {
 
@@ -118,6 +119,19 @@ public interface DataAccessStrategy extends ReadingDataAccessStrategy, RelationR
 	 */
 	<T> boolean updateWithVersion(T instance, Class<T> domainType, Number previousVersion);
 
+	/**
+	 * Upserts the data of a single entity (insert if row for id does not exist, update if it exists). Requires the
+	 * instance to hold an id. Only supported when the dialect supports single-statement upsert.
+	 *
+	 * @param instance the instance to upsert. Must not be {@code null}. Must have an id set.
+	 * @param domainType the type of the instance. Must not be {@code null}.
+	 * @param <T> the type of the instance.
+	 * @return the number of rows affected by the upsert.
+	 * @throws UnsupportedOperationException if the dialect does not support upsert.
+	 * @since 4.x
+	 */
+	<T> int upsert(T instance, Class<? super T> domainType);
+
 	/**
 	 * Deletes a single row identified by the id, from the table identified by the domainType. Does not handle cascading
 	 * deletes.

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DefaultDataAccessStrategy.java

@@ -15,7 +15,7 @@
  */
 package org.springframework.data.jdbc.core.convert;
 
-import static org.springframework.data.jdbc.core.convert.SqlGenerator.*;
+import static org.springframework.data.jdbc.core.convert.SqlGenerator.VERSION_SQL_PARAMETER;
 
 import java.sql.ResultSet;
 import java.sql.SQLException;
@@ -24,8 +24,9 @@
 import java.util.Optional;
 import java.util.stream.Stream;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.jspecify.annotations.Nullable;
-
 import org.springframework.dao.EmptyResultDataAccessException;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
@@ -69,6 +70,8 @@
  */
 public class DefaultDataAccessStrategy implements DataAccessStrategy {
 
+	private final Log logger = LogFactory.getLog(getClass());
+
 	private final SqlGeneratorSource sqlGeneratorSource;
 	private final RelationalMappingContext context;
 	private final JdbcConverter converter;
@@ -179,6 +182,21 @@ public <S> boolean updateWithVersion(S instance, Class<S> domainType, Number pre
 		return true;
 	}
 
+	@Override
+	public <T> int upsert(T instance, Class<? super T> domainType) {
+
+		SqlIdentifierParameterSource parameterSource = sqlParametersFactory.forInsert(instance, domainType, Identifier.empty(),
+				IdValueSource.PROVIDED);
+
+		String statement = sql(domainType).getUpsert(parameterSource.getIdentifiers());
+
+		if (logger.isTraceEnabled()) {
+			logger.trace("Upsert: [%s]".formatted(statement));
+		}
+
+		return operations.update(statement, parameterSource);
+	}
+
 	@Override
 	public void delete(Object id, Class<?> domainType) {
 

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/core/convert/DelegatingDataAccessStrategy.java

@@ -80,6 +80,11 @@ public NamedParameterJdbcOperations getJdbcOperations() {
 		return delegate.insert(insertSubjects, domainType, idValueSource);
 	}
 
+	@Override
+	public <T> int upsert(T instance, Class<? super T> domainType) {
+		return delegate.upsert(instance, domainType);
+	}
+
 	@Override
 	public <S> boolean update(S instance, Class<S> domainType) {
 		return delegate.update(instance, domainType);