DATAJDBC-493 - Avoids deadlocks by acquiring lock on aggregate root table.

Introduces infrastructure to obtain locks and uses them to acquire locks on the table of the aggregate root before deleting references.
Without this lock deletes access non root entities before the aggregate root, which is the opposite order of updates and thus may cause deadlocks.

Original pull request: #196.

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

@@ -27,6 +27,7 @@ import org.springframework.lang.Nullable;
  * Executes an {@link MutableAggregateChange}.
  *
  * @author Jens Schauder
+ * @author Myeonghyeon Lee
  * @since 2.0
  */
 class AggregateChangeExecutor {
@@ -77,6 +78,10 @@ class AggregateChangeExecutor {
 				executionContext.executeDeleteRoot((DbAction.DeleteRoot<?>) action);
 			} else if (action instanceof DbAction.DeleteAllRoot) {
 				executionContext.executeDeleteAllRoot((DbAction.DeleteAllRoot<?>) action);
+			} else if (action instanceof DbAction.AcquireLockRoot) {
+				executionContext.executeAcquireLock((DbAction.AcquireLockRoot<?>) action);
+			} else if (action instanceof DbAction.AcquireLockAllRoot) {
+				executionContext.executeAcquireLockAllRoot((DbAction.AcquireLockAllRoot<?>) action);
 			} else {
 				throw new RuntimeException("unexpected action");
 			}

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

@@ -42,6 +42,7 @@ import org.springframework.data.relational.core.conversion.RelationalEntityVersi
 import org.springframework.data.relational.core.mapping.PersistentPropertyPathExtension;
 import org.springframework.data.relational.core.mapping.RelationalPersistentEntity;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
+import org.springframework.data.relational.core.sql.LockMode;
 import org.springframework.data.util.Pair;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -49,6 +50,7 @@ import org.springframework.util.Assert;
 /**
  * @author Jens Schauder
  * @author Umut Erturk
+ * @author Myeonghyeon Lee
  */
 class JdbcAggregateChangeExecutionContext {
 
@@ -164,6 +166,14 @@ class JdbcAggregateChangeExecutionContext {
 		}
 	}
 
+	<T> void executeAcquireLock(DbAction.AcquireLockRoot<T> acquireLock) {
+		accessStrategy.acquireLockById(acquireLock.getId(), LockMode.PESSIMISTIC_WRITE, acquireLock.getEntityType());
+	}
+
+	<T> void executeAcquireLockAllRoot(DbAction.AcquireLockAllRoot<T> acquireLock) {
+		accessStrategy.acquireLockAll(LockMode.PESSIMISTIC_WRITE, acquireLock.getEntityType());
+	}
+
 	private void add(DbActionExecutionResult result) {
 		results.put(result.getAction(), result);
 	}

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

@@ -50,6 +50,7 @@ import org.springframework.util.Assert;
  * @author Thomas Lang
  * @author Christoph Strobl
  * @author Milan Milanov
+ * @author Myeonghyeon Lee
  */
 public class JdbcAggregateTemplate implements JdbcAggregateOperations {
 

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

@@ -24,6 +24,7 @@ import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
+import org.springframework.data.relational.core.sql.LockMode;
 
 /**
  * Delegates each methods to the {@link DataAccessStrategy}s passed to the constructor in turn until the first that does
@@ -33,6 +34,7 @@ import org.springframework.data.relational.core.mapping.RelationalPersistentProp
  * @author Mark Paluch
  * @author Tyler Van Gorder
  * @author Milan Milanov
+ * @author Myeonghyeon Lee
  * @since 1.1
  */
 public class CascadingDataAccessStrategy implements DataAccessStrategy {
@@ -115,6 +117,24 @@ public class CascadingDataAccessStrategy implements DataAccessStrategy {
 		collectVoid(das -> das.deleteAll(propertyPath));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#acquireLockById(java.lang.Object, org.springframework.data.relational.core.sql.LockMode, java.lang.Class)
+	 */
+	@Override
+	public <T> void acquireLockById(Object id, LockMode lockMode, Class<T> domainType) {
+		collectVoid(das -> das.acquireLockById(id, lockMode, domainType));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#acquireLockAll(org.springframework.data.relational.core.sql.LockMode, java.lang.Class)
+	 */
+	@Override
+	public <T> void acquireLockAll(LockMode lockMode, Class<T> domainType) {
+		collectVoid(das -> das.acquireLockAll(lockMode, domainType));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#count(java.lang.Class)

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

@@ -23,6 +23,7 @@ import org.springframework.data.domain.Sort;
 import org.springframework.data.jdbc.core.JdbcAggregateOperations;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
+import org.springframework.data.relational.core.sql.LockMode;
 import org.springframework.lang.Nullable;
 
 /**
@@ -33,6 +34,7 @@ import org.springframework.lang.Nullable;
  * @author Jens Schauder
  * @author Tyler Van Gorder
  * @author Milan Milanov
+ * @author Myeonghyeon Lee
  */
 public interface DataAccessStrategy extends RelationResolver {
 
@@ -129,6 +131,23 @@ public interface DataAccessStrategy extends RelationResolver {
 	 */
 	void deleteAll(PersistentPropertyPath<RelationalPersistentProperty> propertyPath);
 
+	/**
+	 * Acquire Lock
+	 *
+	 * @param id the id of the entity to load. Must not be {@code null}.
+	 * @param lockMode the lock mode for select. Must not be {@code null}.
+	 * @param domainType the domain type of the entity. Must not be {@code null}.
+	 */
+	<T> void acquireLockById(Object id, LockMode lockMode, Class<T> domainType);
+
+	/**
+	 * Acquire Lock entities of the given domain type.
+	 *
+	 * @param lockMode the lock mode for select. Must not be {@code null}.
+	 * @param domainType the domain type of the entity. Must not be {@code null}.
+	 */
+	<T> void acquireLockAll(LockMode lockMode, Class<T> domainType);
+
 	/**
 	 * Counts the rows in the table representing the given domain type.
 	 *

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

@@ -18,6 +18,8 @@ package org.springframework.data.jdbc.core.convert;
 import static org.springframework.data.jdbc.core.convert.SqlGenerator.*;
 
 import java.sql.JDBCType;
+import java.sql.ResultSet;
+import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashSet;
@@ -25,10 +27,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.function.Predicate;
 
-import org.springframework.dao.DataRetrievalFailureException;
-import org.springframework.dao.EmptyResultDataAccessException;
-import org.springframework.dao.InvalidDataAccessApiUsageException;
-import org.springframework.dao.OptimisticLockingFailureException;
+import org.springframework.dao.*;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.jdbc.support.JdbcUtil;
@@ -41,7 +40,9 @@ 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.IdentifierProcessing;
+import org.springframework.data.relational.core.sql.LockMode;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
+import org.springframework.jdbc.core.ResultSetExtractor;
 import org.springframework.jdbc.core.RowMapper;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
 import org.springframework.jdbc.core.namedparam.SqlParameterSource;
@@ -62,6 +63,7 @@ import org.springframework.util.Assert;
  * @author Tom Hombergs
  * @author Tyler Van Gorder
  * @author Milan Milanov
+ * @author Myeonghyeon Lee
  * @since 1.1
  */
 public class DefaultDataAccessStrategy implements DataAccessStrategy {
@@ -237,6 +239,27 @@ public class DefaultDataAccessStrategy implements DataAccessStrategy {
 				.update(sql(propertyPath.getBaseProperty().getOwner().getType()).createDeleteAllSql(propertyPath));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#acquireLockById(java.lang.Object, org.springframework.data.relational.core.sql.LockMode, java.lang.Class)
+	 */
+	@Override
+	public <T> void acquireLockById(Object id, LockMode lockMode, Class<T> domainType) {
+		String acquireLockByIdSql = sql(domainType).getAcquireLockById(lockMode);
+		SqlIdentifierParameterSource parameter = createIdParameterSource(id, domainType);
+		operations.queryForObject(acquireLockByIdSql, parameter, Object.class);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#acquireLockAll(org.springframework.data.relational.core.sql.LockMode, java.lang.Class)
+	 */
+	@Override
+	public <T> void acquireLockAll(LockMode lockMode, Class<T> domainType) {
+		String acquireLockAllSql = sql(domainType).getAcquireLockAll(lockMode);
+		operations.query(acquireLockAllSql, Collections.emptyMap(), new NoMappingResultSetExtractor());
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#count(java.lang.Class)
@@ -582,4 +605,14 @@ public class DefaultDataAccessStrategy implements DataAccessStrategy {
 			return null;
 		}
 	}
+
+	/**
+	 * The type No mapping result set extractor.
+	 */
+	static class NoMappingResultSetExtractor implements ResultSetExtractor<Object> {
+		@Override
+		public Object extractData(ResultSet resultSet) throws SQLException, DataAccessException {
+			return null;
+		}
+	}
 }

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

@@ -19,6 +19,7 @@ import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.mapping.PersistentPropertyPath;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
+import org.springframework.data.relational.core.sql.LockMode;
 import org.springframework.util.Assert;
 
 /**
@@ -28,6 +29,7 @@ import org.springframework.util.Assert;
  * @author Jens Schauder
  * @author Tyler Van Gorder
  * @author Milan Milanov
+ * @author Myeonghyeon Lee
  * @since 1.1
  */
 public class DelegatingDataAccessStrategy implements DataAccessStrategy {
@@ -107,6 +109,24 @@ public class DelegatingDataAccessStrategy implements DataAccessStrategy {
 		delegate.deleteAll(propertyPath);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#acquireLockById(java.lang.Object, org.springframework.data.relational.core.sql.LockMode, java.lang.Class)
+	 */
+	@Override
+	public <T> void acquireLockById(Object id, LockMode lockMode, Class<T> domainType) {
+		delegate.acquireLockById(id, lockMode, domainType);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#acquireLockAll(org.springframework.data.relational.core.sql.LockMode, java.lang.Class)
+	 */
+	@Override
+	public <T> void acquireLockAll(LockMode lockMode, Class<T> domainType) {
+		delegate.acquireLockAll(lockMode, domainType);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#count(java.lang.Class)

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

@@ -258,6 +258,26 @@ class SqlGenerator {
 		return findOneSql.get();
 	}
 
+	/**
+	 * Create a {@code SELECT count(id) FROM … WHERE :id = … (LOCK CLAUSE)} statement.
+	 *
+	 * @param lockMode Lock clause mode.
+	 * @return the statement as a {@link String}. Guaranteed to be not {@literal null}.
+	 */
+	String getAcquireLockById(LockMode lockMode) {
+		return this.createAcquireLockById(lockMode);
+	}
+
+	/**
+	 * Create a {@code SELECT count(id) FROM … (LOCK CLAUSE)} statement.
+	 *
+	 * @param lockMode Lock clause mode.
+	 * @return the statement as a {@link String}. Guaranteed to be not {@literal null}.
+	 */
+	String getAcquireLockAll(LockMode lockMode) {
+		return this.createAcquireLockAll(lockMode);
+	}
+
 	/**
 	 * Create a {@code INSERT INTO … (…) VALUES(…)} statement.
 	 *
@@ -359,6 +379,33 @@ class SqlGenerator {
 		return render(select);
 	}
 
+	private String createAcquireLockById(LockMode lockMode) {
+
+		Table table = this.getTable();
+
+		Select select = StatementBuilder //
+			.select(getIdColumn()) //
+			.from(table) //
+			.where(getIdColumn().isEqualTo(getBindMarker(ID_SQL_PARAMETER))) //
+			.lock(lockMode) //
+			.build();
+
+		return render(select);
+	}
+
+	private String createAcquireLockAll(LockMode lockMode) {
+
+		Table table = this.getTable();
+
+		Select select = StatementBuilder //
+			.select(getIdColumn()) //
+			.from(table) //
+			.lock(lockMode) //
+			.build();
+
+		return render(select);
+	}
+
 	private String createFindAllSql() {
 		return render(selectBuilder().build());
 	}

spring-data-jdbc/src/main/java/org/springframework/data/jdbc/mybatis/MyBatisDataAccessStrategy.java

@@ -26,6 +26,7 @@ import org.apache.ibatis.session.SqlSession;
 import org.mybatis.spring.SqlSessionTemplate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.dao.EmptyResultDataAccessException;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.jdbc.core.convert.CascadingDataAccessStrategy;
@@ -41,6 +42,7 @@ import org.springframework.data.relational.core.dialect.Dialect;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 import org.springframework.data.relational.core.mapping.RelationalPersistentProperty;
 import org.springframework.data.relational.core.sql.IdentifierProcessing;
+import org.springframework.data.relational.core.sql.LockMode;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
 import org.springframework.util.Assert;
@@ -60,6 +62,7 @@ import org.springframework.util.Assert;
  * @author Mark Paluch
  * @author Tyler Van Gorder
  * @author Milan Milanov
+ * @author Myeonghyeon Lee
  */
 public class MyBatisDataAccessStrategy implements DataAccessStrategy {
 
@@ -248,6 +251,34 @@ public class MyBatisDataAccessStrategy implements DataAccessStrategy {
 		sqlSession().delete(statement, parameter);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#acquireLockById(java.lang.Object, org.springframework.data.relational.core.sql.LockMode, java.lang.Class)
+	 */
+	@Override
+	public <T> void acquireLockById(Object id, LockMode lockMode, Class<T> domainType) {
+		String statement = namespace(domainType) + ".acquireLockById";
+		MyBatisContext parameter = new MyBatisContext(id, null, domainType, Collections.emptyMap());
+
+		long result = sqlSession().selectOne(statement, parameter);
+		if (result < 1) {
+			throw new EmptyResultDataAccessException(
+				String.format("The lock target does not exist. id: %s, statement: %s", id, statement), 1);
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#acquireLockAll(org.springframework.data.relational.core.sql.LockMode, java.lang.Class)
+	 */
+	@Override
+	public <T> void acquireLockAll(LockMode lockMode, Class<T> domainType) {
+		String statement = namespace(domainType) + ".acquireLockAll";
+		MyBatisContext parameter = new MyBatisContext(null, null, domainType, Collections.emptyMap());
+
+		sqlSession().selectOne(statement, parameter);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.jdbc.core.DataAccessStrategy#findById(java.lang.Object, java.lang.Class)

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/NonQuotingDialect.java

@@ -19,6 +19,7 @@ import org.springframework.data.relational.core.dialect.AbstractDialect;
 import org.springframework.data.relational.core.dialect.Dialect;
 import org.springframework.data.relational.core.dialect.HsqlDbDialect;
 import org.springframework.data.relational.core.dialect.LimitClause;
+import org.springframework.data.relational.core.dialect.LockClause;
 import org.springframework.data.relational.core.sql.IdentifierProcessing;
 
 /**
@@ -27,6 +28,7 @@ import org.springframework.data.relational.core.sql.IdentifierProcessing;
  * @author Mark Paluch
  * @author Milan Milanov
  * @author Jens Schauder
+ * @author Myeonghyeon Lee
  */
 public class NonQuotingDialect extends AbstractDialect implements Dialect {
 
@@ -39,6 +41,11 @@ public class NonQuotingDialect extends AbstractDialect implements Dialect {
 		return HsqlDbDialect.INSTANCE.limit();
 	}
 
+	@Override
+	public LockClause lock() {
+		return HsqlDbDialect.INSTANCE.lock();
+	}
+
 	@Override
 	public IdentifierProcessing getIdentifierProcessing() {
 		return IdentifierProcessing.create(new IdentifierProcessing.Quoting(""), IdentifierProcessing.LetterCasing.AS_IS);

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/core/convert/SqlGeneratorUnitTests.java

@@ -46,6 +46,7 @@ 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.Aliased;
+import org.springframework.data.relational.core.sql.LockMode;
 import org.springframework.data.relational.core.sql.SqlIdentifier;
 import org.springframework.data.relational.core.sql.Table;
 
@@ -96,16 +97,45 @@ public class SqlGeneratorUnitTests {
 
 		SoftAssertions softAssertions = new SoftAssertions();
 		softAssertions.assertThat(sql) //
-				.startsWith("SELECT") //
-				.contains("dummy_entity.id1 AS id1,") //
-				.contains("dummy_entity.x_name AS x_name,") //
-				.contains("dummy_entity.x_other AS x_other,") //
-				.contains("ref.x_l1id AS ref_x_l1id") //
-				.contains("ref.x_content AS ref_x_content").contains(" FROM dummy_entity") //
-				.contains("ON ref.dummy_entity = dummy_entity.id1") //
-				.contains("WHERE dummy_entity.id1 = :id") //
-				// 1-N relationships do not get loaded via join
-				.doesNotContain("Element AS elements");
+			.startsWith("SELECT") //
+			.contains("dummy_entity.id1 AS id1,") //
+			.contains("dummy_entity.x_name AS x_name,") //
+			.contains("dummy_entity.x_other AS x_other,") //
+			.contains("ref.x_l1id AS ref_x_l1id") //
+			.contains("ref.x_content AS ref_x_content").contains(" FROM dummy_entity") //
+			.contains("ON ref.dummy_entity = dummy_entity.id1") //
+			.contains("WHERE dummy_entity.id1 = :id") //
+			// 1-N relationships do not get loaded via join
+			.doesNotContain("Element AS elements");
+		softAssertions.assertAll();
+	}
+
+	@Test // DATAJDBC-493
+	public void getAcquireLockById() {
+
+		String sql = sqlGenerator.getAcquireLockById(LockMode.PESSIMISTIC_WRITE);
+
+		SoftAssertions softAssertions = new SoftAssertions();
+		softAssertions.assertThat(sql) //
+			.startsWith("SELECT") //
+			.contains("dummy_entity.id1") //
+			.contains("WHERE dummy_entity.id1 = :id") //
+			.contains("FOR UPDATE") //
+			.doesNotContain("Element AS elements");
+		softAssertions.assertAll();
+	}
+
+	@Test // DATAJDBC-493
+	public void getAcquireLockAll() {
+
+		String sql = sqlGenerator.getAcquireLockAll(LockMode.PESSIMISTIC_WRITE);
+
+		SoftAssertions softAssertions = new SoftAssertions();
+		softAssertions.assertThat(sql) //
+			.startsWith("SELECT") //
+			.contains("dummy_entity.id1") //
+			.contains("FOR UPDATE") //
+			.doesNotContain("Element AS elements");
 		softAssertions.assertAll();
 	}
 

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/repository/JdbcRepositoryConcurrencyIntegrationTests.java

@@ -26,6 +26,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
+import org.springframework.dao.IncorrectUpdateSemanticsDataAccessException;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.jdbc.repository.support.JdbcRepositoryFactory;
 import org.springframework.data.jdbc.testing.DatabaseProfileValueSource;
@@ -34,7 +35,6 @@ import org.springframework.data.repository.CrudRepository;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
 import org.springframework.test.annotation.IfProfileValue;
 import org.springframework.test.annotation.ProfileValueSourceConfiguration;
-import org.springframework.test.context.ActiveProfiles;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.rules.SpringClassRule;
 import org.springframework.test.context.junit4.rules.SpringMethodRule;
@@ -123,6 +123,131 @@ public class JdbcRepositoryConcurrencyIntegrationTests {
 		assertThat(exceptions).isEmpty();
 	}
 
+	@Test // DATAJDBC-493
+	public void updateConcurrencyWithDelete() throws Exception {
+
+		DummyEntity entity = createDummyEntity();
+		entity = repository.save(entity);
+
+		Long targetId = entity.getId();
+		assertThat(targetId).isNotNull();
+
+		List<DummyEntity> concurrencyEntities = createEntityStates(entity);
+
+		TransactionTemplate transactionTemplate = new TransactionTemplate(this.transactionManager);
+
+		List<Exception> exceptions = new CopyOnWriteArrayList<>();
+		CountDownLatch startLatch = new CountDownLatch(concurrencyEntities.size() + 1); // latch for all threads to wait on.
+		CountDownLatch doneLatch = new CountDownLatch(concurrencyEntities.size() + 1); // latch for main thread to wait on until all threads are done.
+
+		// update
+		concurrencyEntities.stream() //
+			.map(e -> new Thread(() -> {
+
+				try {
+
+					startLatch.countDown();
+					startLatch.await();
+
+					transactionTemplate.execute(status -> repository.save(e));
+				} catch (Exception ex) {
+					// When the delete execution is complete, the Update execution throws an IncorrectUpdateSemanticsDataAccessException.
+					if (ex.getCause() instanceof IncorrectUpdateSemanticsDataAccessException) {
+						return;
+					}
+
+					exceptions.add(ex);
+				} finally {
+					doneLatch.countDown();
+				}
+			})) //
+			.forEach(Thread::start);
+
+		// delete
+		new Thread(() -> {
+			try {
+
+				startLatch.countDown();
+				startLatch.await();
+
+				transactionTemplate.execute(status -> {
+					repository.deleteById(targetId);
+					return null;
+				});
+			} catch (Exception ex) {
+				exceptions.add(ex);
+			} finally {
+				doneLatch.countDown();
+			}
+		}).start();
+
+		doneLatch.await();
+
+		assertThat(exceptions).isEmpty();
+		assertThat(repository.findById(entity.id)).isEmpty();
+	}
+
+	@Test // DATAJDBC-493
+	public void updateConcurrencyWithDeleteAll() throws Exception {
+
+		DummyEntity entity = createDummyEntity();
+		entity = repository.save(entity);
+
+		List<DummyEntity> concurrencyEntities = createEntityStates(entity);
+
+		TransactionTemplate transactionTemplate = new TransactionTemplate(this.transactionManager);
+
+		List<Exception> exceptions = new CopyOnWriteArrayList<>();
+		CountDownLatch startLatch = new CountDownLatch(concurrencyEntities.size() + 1); // latch for all threads to wait on.
+		CountDownLatch doneLatch = new CountDownLatch(concurrencyEntities.size() + 1); // latch for main thread to wait on until all threads are done.
+
+		// update
+		concurrencyEntities.stream() //
+			.map(e -> new Thread(() -> {
+
+				try {
+
+					startLatch.countDown();
+					startLatch.await();
+
+					transactionTemplate.execute(status -> repository.save(e));
+				} catch (Exception ex) {
+					// When the delete execution is complete, the Update execution throws an IncorrectUpdateSemanticsDataAccessException.
+					if (ex.getCause() instanceof IncorrectUpdateSemanticsDataAccessException) {
+						return;
+					}
+
+					exceptions.add(ex);
+				} finally {
+					doneLatch.countDown();
+				}
+			})) //
+			.forEach(Thread::start);
+
+		// delete
+		new Thread(() -> {
+			try {
+
+				startLatch.countDown();
+				startLatch.await();
+
+				transactionTemplate.execute(status -> {
+					repository.deleteAll();
+					return null;
+				});
+			} catch (Exception ex) {
+				exceptions.add(ex);
+			} finally {
+				doneLatch.countDown();
+			}
+		}).start();
+
+		doneLatch.await();
+
+		assertThat(exceptions).isEmpty();
+		assertThat(repository.count()).isEqualTo(0);
+	}
+
 	private List<DummyEntity> createEntityStates(DummyEntity entity) {
 
 		List<DummyEntity> concurrencyEntities = new ArrayList<>();

spring-data-jdbc/src/test/java/org/springframework/data/jdbc/testing/AnsiDialect.java

@@ -20,7 +20,9 @@ import lombok.RequiredArgsConstructor;
 import org.springframework.data.relational.core.dialect.AbstractDialect;
 import org.springframework.data.relational.core.dialect.ArrayColumns;
 import org.springframework.data.relational.core.dialect.LimitClause;
+import org.springframework.data.relational.core.dialect.LockClause;
 import org.springframework.data.relational.core.sql.IdentifierProcessing;
+import org.springframework.data.relational.core.sql.LockOptions;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 
@@ -28,6 +30,7 @@ import org.springframework.util.ClassUtils;
  * An SQL dialect for the ANSI SQL standard.
  *
  * @author Milan Milanov
+ * @author Myeonghyeon Lee
  * @since 2.0
  */
 public class AnsiDialect extends AbstractDialect {
@@ -78,6 +81,27 @@ public class AnsiDialect extends AbstractDialect {
 		}
 	};
 
+	private static final LockClause LOCK_CLAUSE = new LockClause() {
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.dialect.LockClause#getLock(LockOptions)
+		 */
+		@Override
+		public String getLock(LockOptions lockOptions) {
+			return "FOR UPDATE";
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.dialect.LimitClause#getClausePosition()
+		 */
+		@Override
+		public Position getClausePosition() {
+			return Position.AFTER_ORDER_BY;
+		}
+	};
+
 	private final AnsiArrayColumns ARRAY_COLUMNS = new AnsiArrayColumns();
 
 	/*
@@ -89,6 +113,15 @@ public class AnsiDialect extends AbstractDialect {
 		return LIMIT_CLAUSE;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.relational.core.dialect.Dialect#lock()
+	 */
+	@Override
+	public LockClause lock() {
+		return LOCK_CLAUSE;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.relational.core.dialect.Dialect#getArraySupport()

spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/DbAction.java

@@ -32,6 +32,7 @@ import org.springframework.lang.Nullable;
  * @author Jens Schauder
  * @author Mark Paluch
  * @author Tyler Van Gorder
+ * @author Myeonghyeon Lee
  */
 public interface DbAction<T> {
 
@@ -319,6 +320,56 @@ public interface DbAction<T> {
 		}
 	}
 
+	/**
+	 * Represents an acquire lock statement for a aggregate root when only the ID is known.
+	 * <p>
+	 *
+	 * @param <T> type of the entity for which this represents a database interaction.
+	 */
+	final class AcquireLockRoot<T> implements DbAction<T> {
+		private final Object id;
+		private final Class<T> entityType;
+
+		public AcquireLockRoot(Object id, Class<T> entityType) {
+			this.id = id;
+			this.entityType = entityType;
+		}
+
+		public Object getId() {
+			return this.id;
+		}
+
+		public Class<T> getEntityType() {
+			return this.entityType;
+		}
+
+		public String toString() {
+			return "DbAction.AcquireLockRoot(id=" + this.getId() + ", entityType=" + this.getEntityType() + ")";
+		}
+	}
+
+	/**
+	 * Represents an acquire lock statement for all entities that that a reachable via a give path from any aggregate root of a
+	 * given type.
+	 *
+	 * @param <T> type of the entity for which this represents a database interaction.
+	 */
+	final class AcquireLockAllRoot<T> implements DbAction<T> {
+		private final Class<T> entityType;
+
+		public AcquireLockAllRoot(Class<T> entityType) {
+			this.entityType = entityType;
+		}
+
+		public Class<T> getEntityType() {
+			return this.entityType;
+		}
+
+		public String toString() {
+			return "DbAction.AcquireLockAllRoot(entityType=" + this.getEntityType() + ")";
+		}
+	}
+
 	/**
 	 * An action depending on another action for providing additional information like the id of a parent entity.
 	 *

spring-data-relational/src/main/java/org/springframework/data/relational/core/conversion/RelationalEntityDeleteWriter.java

@@ -36,6 +36,7 @@ import org.springframework.util.Assert;
  * @author Mark Paluch
  * @author Bastian Wilhelm
  * @author Tyler Van Gorder
+ * @author Myeonghyeon Lee
  */
 public class RelationalEntityDeleteWriter implements EntityWriter<Object, MutableAggregateChange<?>> {
 
@@ -68,12 +69,18 @@ public class RelationalEntityDeleteWriter implements EntityWriter<Object, Mutabl
 
 	private List<DbAction<?>> deleteAll(Class<?> entityType) {
 
-		List<DbAction<?>> actions = new ArrayList<>();
+		List<DbAction<?>> deleteReferencedActions = new ArrayList<>();
 
 		context.findPersistentPropertyPaths(entityType, PersistentProperty::isEntity)
-				.filter(p -> !p.getRequiredLeafProperty().isEmbedded()).forEach(p -> actions.add(new DbAction.DeleteAll<>(p)));
+				.filter(p -> !p.getRequiredLeafProperty().isEmbedded()).forEach(p -> deleteReferencedActions.add(new DbAction.DeleteAll<>(p)));
 
-		Collections.reverse(actions);
+		Collections.reverse(deleteReferencedActions);
+
+		List<DbAction<?>> actions = new ArrayList<>();
+		if (!deleteReferencedActions.isEmpty()) {
+			actions.add(new DbAction.AcquireLockAllRoot<>(entityType));
+		}
+		actions.addAll(deleteReferencedActions);
 
 		DbAction.DeleteAllRoot<?> result = new DbAction.DeleteAllRoot<>(entityType);
 		actions.add(result);
@@ -83,7 +90,14 @@ public class RelationalEntityDeleteWriter implements EntityWriter<Object, Mutabl
 
 	private <T> List<DbAction<?>> deleteRoot(Object id, AggregateChange<T> aggregateChange) {
 
-		List<DbAction<?>> actions = new ArrayList<>(deleteReferencedEntities(id, aggregateChange));
+		List<DbAction<?>> deleteReferencedActions = deleteReferencedEntities(id, aggregateChange);
+
+		List<DbAction<?>> actions = new ArrayList<>();
+		if (!deleteReferencedActions.isEmpty()) {
+			actions.add(new DbAction.AcquireLockRoot<>(id, aggregateChange.getEntityType()));
+		}
+		actions.addAll(deleteReferencedActions);
+
 		actions.add(new DbAction.DeleteRoot<>(id, aggregateChange.getEntityType(), getVersion(aggregateChange)));
 
 		return actions;

spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/AbstractDialect.java

@@ -20,6 +20,8 @@ import lombok.RequiredArgsConstructor;
 import java.util.OptionalLong;
 import java.util.function.Function;
 
+import org.springframework.data.relational.core.sql.LockMode;
+import org.springframework.data.relational.core.sql.LockOptions;
 import org.springframework.data.relational.core.sql.Select;
 import org.springframework.data.relational.core.sql.render.SelectRenderContext;
 
@@ -27,6 +29,7 @@ import org.springframework.data.relational.core.sql.render.SelectRenderContext;
  * Base class for {@link Dialect} implementations.
  *
  * @author Mark Paluch
+ * @author Myeonghyeon Lee
  * @since 1.1
  */
 public abstract class AbstractDialect implements Dialect {
@@ -38,9 +41,31 @@ public abstract class AbstractDialect implements Dialect {
 	@Override
 	public SelectRenderContext getSelectContext() {
 
+		Function<Select, ? extends CharSequence> afterFromTable = getAfterFromTable();
 		Function<Select, ? extends CharSequence> afterOrderBy = getAfterOrderBy();
 
-		return new DialectSelectRenderContext(afterOrderBy);
+		return new DialectSelectRenderContext(afterFromTable, afterOrderBy);
+	}
+
+	/**
+	 * Returns a {@link Function afterFromTable Function}. Typically used for table hint for SQL Server.
+	 *
+	 * @return the {@link Function} called on {@code afterFromTable}.
+	 */
+	protected Function<Select, CharSequence> getAfterFromTable() {
+
+		Function<Select, ? extends CharSequence> afterFromTable = select -> "";
+
+		LockClause lockClause = lock();
+		switch (lockClause.getClausePosition()) {
+
+			case AFTER_FROM_TABLE:
+				afterFromTable = new LockRenderFunction(lockClause);
+
+			default:
+		}
+
+		return afterFromTable.andThen(PrependWithLeadingWhitespace.INSTANCE);
 	}
 
 	/**
@@ -50,21 +75,50 @@ public abstract class AbstractDialect implements Dialect {
 	 */
 	protected Function<Select, CharSequence> getAfterOrderBy() {
 
-		Function<Select, ? extends CharSequence> afterOrderBy;
+		Function<Select, ? extends CharSequence> afterOrderByLimit = getAfterOrderByLimit();
+		Function<Select, ? extends CharSequence> afterOrderByLock = getAfterOrderByLock();
+
+		return select -> {
 
+			StringBuilder afterOrderByBuilder = new StringBuilder();
+			afterOrderByBuilder.append(afterOrderByLimit.apply(select));
+			afterOrderByBuilder.append(afterOrderByLock.apply(select));
+			return afterOrderByBuilder.toString();
+		};
+	}
+
+	private Function<Select, ? extends CharSequence> getAfterOrderByLimit() {
 		LimitClause limit = limit();
 
+		Function<Select, ? extends CharSequence> afterOrderByLimit = select -> "";
+
 		switch (limit.getClausePosition()) {
 
 			case AFTER_ORDER_BY:
-				afterOrderBy = new AfterOrderByLimitRenderFunction(limit);
+				afterOrderByLimit = new AfterOrderByLimitRenderFunction(limit);
 				break;
 
 			default:
 				throw new UnsupportedOperationException(String.format("Clause position %s not supported!", limit));
 		}
 
-		return afterOrderBy.andThen(PrependWithLeadingWhitespace.INSTANCE);
+		return afterOrderByLimit.andThen(PrependWithLeadingWhitespace.INSTANCE);
+	}
+
+	private Function<Select, ? extends CharSequence> getAfterOrderByLock() {
+		LockClause lock = lock();
+
+		Function<Select, ? extends CharSequence> afterOrderByLock = select -> "";
+
+		switch (lock.getClausePosition()) {
+
+			case AFTER_ORDER_BY:
+				afterOrderByLock = new LockRenderFunction(lock);
+
+			default:
+		}
+
+		return afterOrderByLock.andThen(PrependWithLeadingWhitespace.INSTANCE);
 	}
 
 	/**
@@ -72,12 +126,26 @@ public abstract class AbstractDialect implements Dialect {
 	 */
 	class DialectSelectRenderContext implements SelectRenderContext {
 
+		private final Function<Select, ? extends CharSequence> afterFromTable;
 		private final Function<Select, ? extends CharSequence> afterOrderBy;
 
-		DialectSelectRenderContext(Function<Select, ? extends CharSequence> afterOrderBy) {
+		DialectSelectRenderContext(
+				Function<Select, ? extends CharSequence> afterFromTable,
+				Function<Select, ? extends CharSequence> afterOrderBy) {
+
+			this.afterFromTable = afterFromTable;
 			this.afterOrderBy = afterOrderBy;
 		}
 
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.sql.render.SelectRenderContext#afterFromTable()
+		 */
+		@Override
+		public Function<Select, ? extends CharSequence> afterFromTable() {
+			return afterFromTable;
+		}
+
 		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.relational.core.sql.render.SelectRenderContext#afterOrderBy(boolean)
@@ -122,6 +190,31 @@ public abstract class AbstractDialect implements Dialect {
 		}
 	}
 
+	/**
+	 * {@code LOCK} function rendering the {@link LockClause}.
+	 */
+	@RequiredArgsConstructor
+	static class LockRenderFunction implements Function<Select, CharSequence> {
+
+		private final LockClause clause;
+
+		/*
+		 * (non-Javadoc)
+		 * @see java.util.function.Function#apply(java.lang.Object)
+		 */
+		@Override
+		public CharSequence apply(Select select) {
+
+			LockMode lockMode = select.getLockMode();
+
+			if (lockMode == null) {
+				return "";
+			}
+
+			return clause.getLock(new LockOptions(lockMode, select.getFrom()));
+		}
+	}
+
 	/**
 	 * Prepends a non-empty rendering result with a leading whitespace,
 	 */

spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/Dialect.java

@@ -26,6 +26,7 @@ import org.springframework.data.relational.core.sql.render.SelectRenderContext;
  *
  * @author Mark Paluch
  * @author Jens Schauder
+ * @author Myeonghyeon Lee
  * @since 1.1
  */
 public interface Dialect {
@@ -37,6 +38,13 @@ public interface Dialect {
 	 */
 	LimitClause limit();
 
+	/**
+	 * Return the {@link LockClause} used by this dialect.
+	 *
+	 * @return the {@link LockClause} used by this dialect.
+	 */
+	LockClause lock();
+
 	/**
 	 * Returns the array support object that describes how array-typed columns are supported by this dialect.
 	 *

spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/H2Dialect.java

@@ -20,6 +20,7 @@ import lombok.RequiredArgsConstructor;
 import org.springframework.data.relational.core.sql.IdentifierProcessing;
 import org.springframework.data.relational.core.sql.IdentifierProcessing.LetterCasing;
 import org.springframework.data.relational.core.sql.IdentifierProcessing.Quoting;
+import org.springframework.data.relational.core.sql.LockOptions;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 
@@ -27,6 +28,7 @@ import org.springframework.util.ClassUtils;
  * An SQL dialect for H2.
  *
  * @author Mark Paluch
+ * @author Myeonghyeon Lee
  * @since 2.0
  */
 public class H2Dialect extends AbstractDialect {
@@ -77,6 +79,27 @@ public class H2Dialect extends AbstractDialect {
 		}
 	};
 
+	private static final LockClause LOCK_CLAUSE = new LockClause() {
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.dialect.LockClause#getLock(LockOptions)
+		 */
+		@Override
+		public String getLock(LockOptions lockOptions) {
+			return "FOR UPDATE";
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.dialect.LockClause#getClausePosition()
+		 */
+		@Override
+		public Position getClausePosition() {
+			return Position.AFTER_ORDER_BY;
+		}
+	};
+
 	private final H2ArrayColumns ARRAY_COLUMNS = new H2ArrayColumns();
 
 	/*
@@ -88,6 +111,15 @@ public class H2Dialect extends AbstractDialect {
 		return LIMIT_CLAUSE;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.relational.core.dialect.Dialect#lock()
+	 */
+	@Override
+	public LockClause lock() {
+		return LOCK_CLAUSE;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.relational.core.dialect.Dialect#getArraySupport()

spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/HsqlDbDialect.java

@@ -15,10 +15,13 @@
  */
 package org.springframework.data.relational.core.dialect;
 
+import org.springframework.data.relational.core.sql.LockOptions;
+
 /**
  * A {@link Dialect} for HsqlDb.
  *
  * @author Jens Schauder
+ * @author Myeonghyeon Lee
  */
 public class HsqlDbDialect extends AbstractDialect {
 
@@ -31,6 +34,11 @@ public class HsqlDbDialect extends AbstractDialect {
 		return LIMIT_CLAUSE;
 	}
 
+	@Override
+	public LockClause lock() {
+		return LOCK_CLAUSE;
+	}
+
 	private static final LimitClause LIMIT_CLAUSE = new LimitClause() {
 
 		@Override
@@ -53,4 +61,17 @@ public class HsqlDbDialect extends AbstractDialect {
 			return Position.AFTER_ORDER_BY;
 		}
 	};
+
+	private static final LockClause LOCK_CLAUSE = new LockClause() {
+
+		@Override
+		public String getLock(LockOptions lockOptions) {
+			return "FOR UPDATE";
+		}
+
+		@Override
+		public Position getClausePosition() {
+			return Position.AFTER_ORDER_BY;
+		}
+	};
 }

spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/LockClause.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2020 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.relational.core.dialect;
+
+import org.springframework.data.relational.core.sql.LockOptions;
+
+/**
+ * A clause representing Dialect-specific {@code LOCK}.
+ *
+ * @author Myeonghyeon Lee
+ * @since 2.0
+ */
+public interface LockClause {
+
+	/**
+	 * Returns the {@code LOCK} clause to lock results.
+	 *
+	 * @param lockOptions contains the lock mode to apply.
+	 * @return rendered lock clause.
+	 */
+	String getLock(LockOptions lockOptions);
+
+	/**
+	 * Returns the {@link Position} where to apply the {@link #getLock(LockOptions) clause}.
+	 */
+	Position getClausePosition();
+
+	/**
+	 * Enumeration of where to render the clause within the SQL statement.
+	 */
+	enum Position {
+		/**
+		 * Append the clause after from table.
+		 */
+		AFTER_FROM_TABLE,
+
+		/**
+		 * Append the clause at the end of the statement.
+		 */
+		AFTER_ORDER_BY
+	}
+}

spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/MySqlDialect.java

@@ -19,12 +19,14 @@ import org.springframework.data.relational.core.sql.IdentifierProcessing;
 import org.springframework.data.relational.core.sql.IdentifierProcessing.LetterCasing;
 import org.springframework.data.relational.core.sql.IdentifierProcessing.Quoting;
 import org.springframework.util.Assert;
+import org.springframework.data.relational.core.sql.LockOptions;
 
 /**
  * A SQL dialect for MySQL.
  *
  * @author Mark Paluch
  * @author Jens Schauder
+ * @author Myeonghyeon Lee
  * @since 1.1
  */
 public class MySqlDialect extends AbstractDialect {
@@ -102,6 +104,37 @@ public class MySqlDialect extends AbstractDialect {
 		}
 	};
 
+	private static final LockClause LOCK_CLAUSE = new LockClause() {
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.dialect.LockClause#getLock(LockOptions)
+		 */
+		@Override
+		public String getLock(LockOptions lockOptions) {
+			switch (lockOptions.getLockMode()) {
+
+				case PESSIMISTIC_WRITE:
+					return "FOR UPDATE";
+
+				case PESSIMISTIC_READ:
+					return "LOCK IN SHARE MODE";
+
+				default:
+					return "";
+			}
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.dialect.LockClause#getClausePosition()
+		 */
+		@Override
+		public Position getClausePosition() {
+			return Position.AFTER_ORDER_BY;
+		}
+	};
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.relational.core.dialect.Dialect#limit()
@@ -111,6 +144,15 @@ public class MySqlDialect extends AbstractDialect {
 		return LIMIT_CLAUSE;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.relational.core.dialect.Dialect#lock()
+	 */
+	@Override
+	public LockClause lock() {
+		return LOCK_CLAUSE;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.relational.core.dialect.Dialect#getIdentifierProcessing()

spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/PostgresDialect.java

@@ -17,16 +17,22 @@ package org.springframework.data.relational.core.dialect;
 
 import lombok.RequiredArgsConstructor;
 
+import org.springframework.data.relational.core.sql.From;
 import org.springframework.data.relational.core.sql.IdentifierProcessing;
 import org.springframework.data.relational.core.sql.IdentifierProcessing.LetterCasing;
 import org.springframework.data.relational.core.sql.IdentifierProcessing.Quoting;
+import org.springframework.data.relational.core.sql.LockOptions;
+import org.springframework.data.relational.core.sql.Table;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 
+import java.util.List;
+
 /**
  * An SQL dialect for Postgres.
  *
  * @author Mark Paluch
+ * @author Myeonghyeon Lee
  * @since 1.1
  */
 public class PostgresDialect extends AbstractDialect {
@@ -88,6 +94,17 @@ public class PostgresDialect extends AbstractDialect {
 		return LIMIT_CLAUSE;
 	}
 
+	private final PostgresLockClause LOCK_CLAUSE = new PostgresLockClause(this.getIdentifierProcessing());
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.relational.core.dialect.Dialect#lock()
+	 */
+	@Override
+	public LockClause lock() {
+		return LOCK_CLAUSE;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.relational.core.dialect.Dialect#getArraySupport()
@@ -97,6 +114,50 @@ public class PostgresDialect extends AbstractDialect {
 		return ARRAY_COLUMNS;
 	}
 
+	static class PostgresLockClause implements LockClause {
+		private final IdentifierProcessing identifierProcessing;
+
+		PostgresLockClause(IdentifierProcessing identifierProcessing) {
+			this.identifierProcessing = identifierProcessing;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.dialect.LockClause#getLock(LockOptions)
+		 */
+		@Override
+		public String getLock(LockOptions lockOptions) {
+
+			List<Table> tables = lockOptions.getFrom().getTables();
+			if (tables.isEmpty()) {
+				return "";
+			}
+
+			String tableName = tables.get(0).getName().toSql(this.identifierProcessing);
+
+			switch (lockOptions.getLockMode()) {
+
+				case PESSIMISTIC_WRITE:
+					return "FOR UPDATE OF " + tableName;
+
+				case PESSIMISTIC_READ:
+					return "FOR SHARE OF " + tableName;
+
+				default:
+					return "";
+			}
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.dialect.LockClause#getClausePosition()
+		 */
+		@Override
+		public Position getClausePosition() {
+			return Position.AFTER_ORDER_BY;
+		}
+	};
+
 	@RequiredArgsConstructor
 	static class PostgresArrayColumns implements ArrayColumns {
 

spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/SqlServerDialect.java

@@ -15,6 +15,7 @@
  */
 package org.springframework.data.relational.core.dialect;
 
+import org.springframework.data.relational.core.sql.LockOptions;
 import org.springframework.data.relational.core.sql.render.SelectRenderContext;
 import org.springframework.data.util.Lazy;
 
@@ -22,6 +23,7 @@ import org.springframework.data.util.Lazy;
  * An SQL dialect for Microsoft SQL Server.
  *
  * @author Mark Paluch
+ * @author Myeonghyeon Lee
  * @since 1.1
  */
 public class SqlServerDialect extends AbstractDialect {
@@ -72,8 +74,39 @@ public class SqlServerDialect extends AbstractDialect {
 		}
 	};
 
+	private static final LockClause LOCK_CLAUSE = new LockClause() {
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.dialect.LockClause#getLimit(LockOptions)
+		 */
+		@Override
+		public String getLock(LockOptions lockOptions) {
+			switch (lockOptions.getLockMode()) {
+
+				case PESSIMISTIC_WRITE:
+					return "WITH (UPDLOCK, ROWLOCK)";
+
+				case PESSIMISTIC_READ:
+					return "WITH (HOLDLOCK, ROWLOCK)";
+
+				default:
+					return "";
+			}
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.dialect.LimitClause#getClausePosition()
+		 */
+		@Override
+		public Position getClausePosition() {
+			return Position.AFTER_FROM_TABLE;
+		}
+	};
+
 	private final Lazy<SelectRenderContext> selectRenderContext = Lazy
-			.of(() -> new SqlServerSelectRenderContext(getAfterOrderBy()));
+			.of(() -> new SqlServerSelectRenderContext(getAfterFromTable(), getAfterOrderBy()));
 
 	/*
 	 * (non-Javadoc)
@@ -84,6 +117,15 @@ public class SqlServerDialect extends AbstractDialect {
 		return LIMIT_CLAUSE;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.relational.core.dialect.Dialect#lock()
+	 */
+	@Override
+	public LockClause lock() {
+		return LOCK_CLAUSE;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.relational.core.dialect.Dialect#getLikeEscaper()

spring-data-relational/src/main/java/org/springframework/data/relational/core/dialect/SqlServerSelectRenderContext.java

@@ -28,6 +28,7 @@ import org.springframework.data.relational.core.sql.render.SelectRenderContext;
  * </ul>
  *
  * @author Mark Paluch
+ * @author Myeonghyeon Lee
  */
 public class SqlServerSelectRenderContext implements SelectRenderContext {
 
@@ -36,14 +37,20 @@ public class SqlServerSelectRenderContext implements SelectRenderContext {
 	private static final String SYNTHETIC_SELECT_LIST = ", ROW_NUMBER() over (ORDER BY (SELECT 1)) AS "
 			+ SYNTHETIC_ORDER_BY_FIELD;
 
+	private final Function<Select, CharSequence> afterFromTable;
 	private final Function<Select, CharSequence> afterOrderBy;
 
 	/**
 	 * Creates a new {@link SqlServerSelectRenderContext}.
 	 *
+	 * @param afterFromTable the delegate {@code afterFromTable} function.
 	 * @param afterOrderBy the delegate {@code afterOrderBy} function.
 	 */
-	protected SqlServerSelectRenderContext(Function<Select, CharSequence> afterOrderBy) {
+	protected SqlServerSelectRenderContext(
+			Function<Select, CharSequence> afterFromTable,
+			Function<Select, CharSequence> afterOrderBy) {
+
+		this.afterFromTable = afterFromTable;
 		this.afterOrderBy = afterOrderBy;
 	}
 
@@ -60,6 +67,12 @@ public class SqlServerSelectRenderContext implements SelectRenderContext {
 		};
 	}
 
+	@Override
+	public Function<Select, ? extends CharSequence> afterFromTable() {
+
+		return afterFromTable;
+	}
+
 	@Override
 	public Function<Select, ? extends CharSequence> afterOrderBy(boolean hasOrderBy) {
 

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/DefaultSelect.java

@@ -27,6 +27,7 @@ import org.springframework.util.Assert;
  * Default {@link Select} implementation.
  *
  * @author Mark Paluch
+ * @author Myeonghyeon Lee
  * @since 1.1
  */
 class DefaultSelect implements Select {
@@ -39,9 +40,10 @@ class DefaultSelect implements Select {
 	private final List<Join> joins;
 	private final @Nullable Where where;
 	private final List<OrderByField> orderBy;
+	private final @Nullable LockMode lockMode;
 
 	DefaultSelect(boolean distinct, List<Expression> selectList, List<Table> from, long limit, long offset,
-			List<Join> joins, @Nullable Condition where, List<OrderByField> orderBy) {
+			List<Join> joins, @Nullable Condition where, List<OrderByField> orderBy, @Nullable LockMode lockMode) {
 
 		this.distinct = distinct;
 		this.selectList = new SelectList(new ArrayList<>(selectList));
@@ -51,6 +53,16 @@ class DefaultSelect implements Select {
 		this.joins = new ArrayList<>(joins);
 		this.orderBy = Collections.unmodifiableList(new ArrayList<>(orderBy));
 		this.where = where != null ? new Where(where) : null;
+		this.lockMode = lockMode;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.relational.core.sql.Select#getFrom()
+	 */
+	@Override
+	public From getFrom() {
+		return this.from;
 	}
 
 	/*
@@ -85,6 +97,16 @@ class DefaultSelect implements Select {
 		return distinct;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.relational.core.sql.Select#getLockMode()
+	 */
+	@Nullable
+	@Override
+	public LockMode getLockMode() {
+		return lockMode;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.relational.core.sql.Visitable#visit(org.springframework.data.relational.core.sql.Visitor)

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/DefaultSelectBuilder.java

@@ -31,6 +31,7 @@ import org.springframework.lang.Nullable;
  *
  * @author Mark Paluch
  * @author Jens Schauder
+ * @author Myeonghyeon Lee
  * @since 1.1
  */
 class DefaultSelectBuilder implements SelectBuilder, SelectAndFrom, SelectFromAndJoin, SelectWhereAndOr {
@@ -43,6 +44,7 @@ class DefaultSelectBuilder implements SelectBuilder, SelectAndFrom, SelectFromAn
 	private List<Join> joins = new ArrayList<>();
 	private @Nullable Condition where;
 	private List<OrderByField> orderBy = new ArrayList<>();
+	private @Nullable LockMode lockMode;
 
 	/*
 	 * (non-Javadoc)
@@ -265,13 +267,23 @@ class DefaultSelectBuilder implements SelectBuilder, SelectAndFrom, SelectFromAn
 		return this;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.relational.core.sql.SelectBuilder.SelectLock#lock(org.springframework.data.relational.core.sql.LockMode)
+	 */
+	@Override
+	public SelectLock lock(LockMode lockMode) {
+		this.lockMode = lockMode;
+		return this;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.relational.core.sql.SelectBuilder.BuildSelect#build()
 	 */
 	@Override
 	public Select build() {
-		DefaultSelect select = new DefaultSelect(distinct, selectList, from, limit, offset, joins, where, orderBy);
+		DefaultSelect select = new DefaultSelect(distinct, selectList, from, limit, offset, joins, where, orderBy, lockMode);
 		SelectValidator.validate(select);
 		return select;
 	}
@@ -448,6 +460,16 @@ class DefaultSelectBuilder implements SelectBuilder, SelectAndFrom, SelectFromAn
 			return selectBuilder.offset(offset);
 		}
 
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.relational.core.sql.SelectBuilder.SelectLock#lock(org.springframework.data.relational.core.sql.LockMode)
+		 */
+		@Override
+		public SelectLock lock(LockMode lockMode) {
+			selectBuilder.join(finishJoin());
+			return selectBuilder.lock(lockMode);
+		}
+
 		/*
 		 * (non-Javadoc)
 		 * @see org.springframework.data.relational.core.sql.SelectBuilder.BuildSelect#build()

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/From.java

@@ -16,6 +16,7 @@
 package org.springframework.data.relational.core.sql;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 
 import org.springframework.util.StringUtils;
@@ -38,7 +39,11 @@ public class From extends AbstractSegment {
 
 		super(tables.toArray(new Table[] {}));
 
-		this.tables = tables;
+		this.tables = Collections.unmodifiableList(tables);
+	}
+
+	public List<Table> getTables() {
+		return this.tables;
 	}
 
 	/*

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/LockMode.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2020 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.relational.core.sql;
+
+/**
+ * Lock Mode Types of SELECT statements.
+ *
+ * @author Myeonghyeon Lee
+ * @since 2.0
+ */
+public enum LockMode {
+	PESSIMISTIC_READ,
+	PESSIMISTIC_WRITE
+}

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/LockOptions.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2020 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.relational.core.sql;
+
+/**
+ * LockOptions has a LOCK option to apply to the Select statement.
+ *
+ * @author Myeonghyeon Lee
+ * @since 2.0
+ */
+public class LockOptions {
+	private final LockMode lockMode;
+	private final From from;
+
+	public LockOptions(LockMode lockMode, From  from) {
+		this.lockMode = lockMode;
+		this.from = from;
+	}
+
+	public LockMode getLockMode() {
+		return this.lockMode;
+	}
+
+	public From getFrom() {
+		return this.from;
+	}
+}

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/Select.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.relational.core.sql;
 
+import org.springframework.lang.Nullable;
+
 import java.util.List;
 import java.util.OptionalLong;
 
@@ -30,6 +32,7 @@ import java.util.OptionalLong;
  * </ol>
  *
  * @author Mark Paluch
+ * @author Myeonghyeon Lee
  * @since 1.1
  * @see StatementBuilder
  * @see SelectBuilder
@@ -46,6 +49,8 @@ public interface Select extends Segment, Visitable {
 		return new DefaultSelectBuilder();
 	}
 
+	From getFrom();
+
 	/**
 	 * @return the {@link List} of {@link OrderByField ORDER BY} fields.
 	 */
@@ -71,4 +76,7 @@ public interface Select extends Segment, Visitable {
 	 * @return
 	 */
 	boolean isDistinct();
+
+	@Nullable
+	LockMode getLockMode();
 }

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/SelectBuilder.java

@@ -22,6 +22,7 @@ import java.util.Collection;
  *
  * @author Mark Paluch
  * @author Jens Schauder
+ * @author Myeonghyeon Lee
  * @since 1.1
  * @see StatementBuilder
  */
@@ -211,9 +212,9 @@ public interface SelectBuilder {
 	}
 
 	/**
-	 * Builder exposing {@code FROM}, {@code JOIN}, {@code WHERE} and {@code LIMIT/OFFSET} methods.
+	 * Builder exposing {@code FROM}, {@code JOIN}, {@code WHERE}, {@code LIMIT/OFFSET} and {@code LOCK} methods.
 	 */
-	interface SelectFromAndOrderBy extends SelectFrom, SelectOrdered, SelectLimitOffset, BuildSelect {
+	interface SelectFromAndOrderBy extends SelectFrom, SelectOrdered, SelectLimitOffset, SelectLock, BuildSelect {
 
 		@Override
 		SelectFromAndOrderBy limitOffset(long limit, long offset);
@@ -247,9 +248,10 @@ public interface SelectBuilder {
 	}
 
 	/**
-	 * Builder exposing {@code FROM}, {@code JOIN}, {@code WHERE} and {@code LIMIT/OFFSET} methods.
+	 * Builder exposing {@code FROM}, {@code JOIN}, {@code WHERE}, {@code LIMIT/OFFSET} and {@code LOCK} methods.
 	 */
-	interface SelectFromAndJoin extends SelectFromAndOrderBy, BuildSelect, SelectJoin, SelectWhere, SelectLimitOffset {
+	interface SelectFromAndJoin
+		extends SelectFromAndOrderBy, BuildSelect, SelectJoin, SelectWhere, SelectLimitOffset, SelectLock {
 
 		/**
 		 * Declare a {@link Table} to {@code SELECT … FROM}. Multiple calls to this or other {@code from} methods keep
@@ -315,10 +317,10 @@ public interface SelectBuilder {
 	}
 
 	/**
-	 * Builder exposing {@code FROM}, {@code WHERE}, {@code LIMIT/OFFSET}, and JOIN {@code AND} continuation methods.
+	 * Builder exposing {@code FROM}, {@code WHERE}, {@code LIMIT/OFFSET}, JOIN {@code AND} and {@code LOCK} continuation methods.
 	 */
 	interface SelectFromAndJoinCondition
-			extends BuildSelect, SelectJoin, SelectWhere, SelectOnCondition, SelectLimitOffset {
+			extends BuildSelect, SelectJoin, SelectWhere, SelectOnCondition, SelectLimitOffset, SelectLock {
 
 		/**
 		 * Apply {@code limit} and {@code offset} parameters to the select statement. To read the first 20 rows from start
@@ -380,9 +382,9 @@ public interface SelectBuilder {
 	}
 
 	/**
-	 * Builder exposing {@code ORDER BY} methods.
+	 * Builder exposing {@code ORDER BY} and {@code LOCK} methods.
 	 */
-	interface SelectOrdered extends BuildSelect {
+	interface SelectOrdered extends SelectLock, BuildSelect {
 
 		/**
 		 * Add one or more {@link Column columns} to order by.
@@ -410,9 +412,9 @@ public interface SelectBuilder {
 	}
 
 	/**
-	 * Interface exposing {@code WHERE} methods.
+	 * Interface exposing {@code WHERE}, {@code LOCK} methods.
 	 */
-	interface SelectWhere extends SelectOrdered, BuildSelect {
+	interface SelectWhere extends SelectOrdered, SelectLock, BuildSelect {
 
 		/**
 		 * Apply a {@code WHERE} clause.
@@ -428,7 +430,7 @@ public interface SelectBuilder {
 	/**
 	 * Interface exposing {@code AND}/{@code OR} combinator methods for {@code WHERE} {@link Condition}s.
 	 */
-	interface SelectWhereAndOr extends SelectOrdered, BuildSelect {
+	interface SelectWhereAndOr extends SelectOrdered, SelectLock, BuildSelect {
 
 		/**
 		 * Combine the previous {@code WHERE} {@link Condition} using {@code AND}.
@@ -452,7 +454,7 @@ public interface SelectBuilder {
 	/**
 	 * Interface exposing {@code JOIN} methods.
 	 */
-	interface SelectJoin extends BuildSelect {
+	interface SelectJoin extends SelectLock, BuildSelect {
 
 		/**
 		 * Declare a {@code JOIN} {@code table}.
@@ -518,7 +520,7 @@ public interface SelectBuilder {
 	/**
 	 * Builder exposing JOIN and {@code JOIN … ON} continuation methods.
 	 */
-	interface SelectOnCondition extends SelectJoin, BuildSelect {
+	interface SelectOnCondition extends SelectJoin, SelectLock, BuildSelect {
 
 		/**
 		 * Declare an additional source column in the {@code JOIN}.
@@ -530,6 +532,20 @@ public interface SelectBuilder {
 		SelectOnConditionComparison and(Expression column);
 	}
 
+	/**
+	 * Lock methods.
+	 */
+	interface SelectLock extends BuildSelect {
+
+		/**
+		 * Apply lock to read.
+		 *
+		 * @param lockMode lockMode to read.
+		 * @return {@code this} builder.
+		 */
+		SelectLock lock(LockMode lockMode);
+	}
+
 	/**
 	 * Interface exposing the {@link Select} build method.
 	 */

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/render/SelectRenderContext.java

@@ -25,6 +25,7 @@ import org.springframework.data.relational.core.sql.Select;
  * element without further whitespace processing. Hooks are responsible for adding required surrounding whitespaces.
  *
  * @author Mark Paluch
+ * @author Myeonghyeon Lee
  * @since 1.1
  */
 public interface SelectRenderContext {
@@ -39,6 +40,16 @@ public interface SelectRenderContext {
 		return select -> "";
 	}
 
+	/**
+	 * Customization hook: Rendition of a part after {@code FROM} table.
+	 * Renders an empty string by default.
+	 *
+	 * @return render {@link Function} invoked after rendering {@code FROM} table.
+	 */
+	default Function<Select, ? extends CharSequence> afterFromTable() {
+		return select -> "";
+	}
+
 	/**
 	 * Customization hook: Rendition of a part after {@code ORDER BY}. The rendering function is called always, regardless
 	 * whether {@code ORDER BY} exists or not. Renders an empty string by default.

spring-data-relational/src/main/java/org/springframework/data/relational/core/sql/render/SelectStatementVisitor.java

@@ -28,6 +28,7 @@ import org.springframework.data.relational.core.sql.Where;
  *
  * @author Mark Paluch
  * @author Jens Schauder
+ * @author Myeonghyeon Lee
  * @since 1.1
  */
 class SelectStatementVisitor extends DelegatingVisitor implements PartRenderer {
@@ -125,6 +126,8 @@ class SelectStatementVisitor extends DelegatingVisitor implements PartRenderer {
 				builder.append(" FROM ").append(from);
 			}
 
+			builder.append(selectRenderContext.afterFromTable().apply(select));
+
 			if (join.length() != 0) {
 				builder.append(' ').append(join);
 			}

spring-data-relational/src/test/java/org/springframework/data/relational/core/conversion/RelationalEntityDeleteWriterUnitTests.java

@@ -16,26 +16,23 @@
 package org.springframework.data.relational.core.conversion;
 
 import lombok.Data;
-
-import java.util.ArrayList;
-import java.util.List;
-
 import org.assertj.core.api.Assertions;
 import org.assertj.core.groups.Tuple;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.springframework.data.annotation.Id;
-import org.springframework.data.relational.core.conversion.DbAction.Delete;
-import org.springframework.data.relational.core.conversion.DbAction.DeleteAll;
-import org.springframework.data.relational.core.conversion.DbAction.DeleteAllRoot;
-import org.springframework.data.relational.core.conversion.DbAction.DeleteRoot;
+import org.springframework.data.relational.core.conversion.DbAction.*;
 import org.springframework.data.relational.core.mapping.RelationalMappingContext;
 
+import java.util.ArrayList;
+import java.util.List;
+
 /**
  * Unit tests for the {@link org.springframework.data.relational.core.conversion.RelationalEntityDeleteWriter}
  *
  * @author Jens Schauder
+ * @author Myeonghyeon Lee
  */
 @RunWith(MockitoJUnitRunner.class)
 public class RelationalEntityDeleteWriterUnitTests {
@@ -54,12 +51,27 @@ public class RelationalEntityDeleteWriterUnitTests {
 		Assertions.assertThat(extractActions(aggregateChange))
 				.extracting(DbAction::getClass, DbAction::getEntityType, DbActionTestSupport::extractPath) //
 				.containsExactly( //
+						Tuple.tuple(AcquireLockRoot.class, SomeEntity.class, ""), //
 						Tuple.tuple(Delete.class, YetAnother.class, "other.yetAnother"), //
 						Tuple.tuple(Delete.class, OtherEntity.class, "other"), //
 						Tuple.tuple(DeleteRoot.class, SomeEntity.class, "") //
 				);
 	}
 
+	@Test // DATAJDBC-493
+	public void deleteDeletesTheEntityAndNoReferencedEntities() {
+
+		SingleEntity entity = new SingleEntity(23L);
+
+		MutableAggregateChange<SingleEntity> aggregateChange = MutableAggregateChange.forDelete(SingleEntity.class, entity);
+
+		converter.write(entity.id, aggregateChange);
+
+		Assertions.assertThat(extractActions(aggregateChange))
+			.extracting(DbAction::getClass, DbAction::getEntityType, DbActionTestSupport::extractPath) //
+			.containsExactly(Tuple.tuple(DeleteRoot.class, SingleEntity.class, ""));
+	}
+
 	@Test // DATAJDBC-188
 	public void deleteAllDeletesAllEntitiesAndReferencedEntities() {
 
@@ -70,12 +82,25 @@ public class RelationalEntityDeleteWriterUnitTests {
 		Assertions.assertThat(extractActions(aggregateChange))
 				.extracting(DbAction::getClass, DbAction::getEntityType, DbActionTestSupport::extractPath) //
 				.containsExactly( //
+						Tuple.tuple(AcquireLockAllRoot.class, SomeEntity.class, ""), //
 						Tuple.tuple(DeleteAll.class, YetAnother.class, "other.yetAnother"), //
 						Tuple.tuple(DeleteAll.class, OtherEntity.class, "other"), //
 						Tuple.tuple(DeleteAllRoot.class, SomeEntity.class, "") //
 				);
 	}
 
+	@Test // DATAJDBC-493
+	public void deleteAllDeletesAllEntitiesAndNoReferencedEntities() {
+
+		MutableAggregateChange<SingleEntity> aggregateChange = MutableAggregateChange.forDelete(SingleEntity.class, null);
+
+		converter.write(null, aggregateChange);
+
+		Assertions.assertThat(extractActions(aggregateChange))
+			.extracting(DbAction::getClass, DbAction::getEntityType, DbActionTestSupport::extractPath) //
+			.containsExactly(Tuple.tuple(DeleteAllRoot.class, SingleEntity.class, ""));
+	}
+
 	private List<DbAction<?>> extractActions(MutableAggregateChange<?> aggregateChange) {
 
 		List<DbAction<?>> actions = new ArrayList<>();
@@ -103,4 +128,10 @@ public class RelationalEntityDeleteWriterUnitTests {
 	private class YetAnother {
 		@Id final Long id;
 	}
+
+	@Data
+	private class SingleEntity {
+		@Id final Long id;
+		String name;
+	}
 }

spring-data-relational/src/test/java/org/springframework/data/relational/core/dialect/HsqlDbDialectUnitTests.java

@@ -15,14 +15,19 @@
  */
 package org.springframework.data.relational.core.dialect;
 
-import static org.assertj.core.api.Assertions.*;
-
 import org.junit.Test;
+import org.springframework.data.relational.core.sql.From;
+import org.springframework.data.relational.core.sql.LockMode;
+import org.springframework.data.relational.core.sql.LockOptions;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
 
 /**
  * Unit tests for the {@link HsqlDbDialect}.
- * 
+ *
  * @author Jens Schauder
+ * @author Myeonghyeon Lee
  */
 public class HsqlDbDialectUnitTests {
 
@@ -66,4 +71,15 @@ public class HsqlDbDialectUnitTests {
 
 		assertThat(abcQuoted).isEqualTo("\"abc\"");
 	}
+
+	@Test // DATAJDBC-498
+	public void shouldRenderLock() {
+
+		LockClause limit = HsqlDbDialect.INSTANCE.lock();
+		From from = mock(From.class);
+		LockOptions lockOptions = new LockOptions(LockMode.PESSIMISTIC_WRITE, from);
+
+		assertThat(limit.getLock(lockOptions)).isEqualTo("FOR UPDATE");
+		assertThat(limit.getClausePosition()).isEqualTo(LockClause.Position.AFTER_ORDER_BY);
+	}
 }

spring-data-relational/src/test/java/org/springframework/data/relational/core/dialect/MySqlDialectRenderingUnitTests.java

@@ -20,6 +20,7 @@ import static org.assertj.core.api.Assertions.*;
 import org.junit.Before;
 import org.junit.Test;
 
+import org.springframework.data.relational.core.sql.LockMode;
 import org.springframework.data.relational.core.sql.Select;
 import org.springframework.data.relational.core.sql.StatementBuilder;
 import org.springframework.data.relational.core.sql.Table;
@@ -31,6 +32,7 @@ import org.springframework.data.relational.core.sql.render.SqlRenderer;
  *
  * @author Mark Paluch
  * @author Jens Schauder
+ * @author Myeonghyeon Lee
  */
 public class MySqlDialectRenderingUnitTests {
 
@@ -73,4 +75,52 @@ public class MySqlDialectRenderingUnitTests {
 
 		assertThat(sql).isEqualTo("SELECT foo.* FROM foo LIMIT 20, 10");
 	}
+
+	@Test // DATAJDBC-498
+	public void shouldRenderSelectWithLockWrite() {
+
+		Table table = Table.create("foo");
+		LockMode lockMode = LockMode.PESSIMISTIC_WRITE;
+		Select select = StatementBuilder.select(table.asterisk()).from(table).lock(lockMode).build();
+
+		String sql = SqlRenderer.create(factory.createRenderContext()).render(select);
+
+		assertThat(sql).isEqualTo("SELECT foo.* FROM foo FOR UPDATE");
+	}
+
+	@Test // DATAJDBC-498
+	public void shouldRenderSelectWithLockRead() {
+
+		Table table = Table.create("foo");
+		LockMode lockMode = LockMode.PESSIMISTIC_READ;
+		Select select = StatementBuilder.select(table.asterisk()).from(table).lock(lockMode).build();
+
+		String sql = SqlRenderer.create(factory.createRenderContext()).render(select);
+
+		assertThat(sql).isEqualTo("SELECT foo.* FROM foo LOCK IN SHARE MODE");
+	}
+
+	@Test // DATAJDBC-498
+	public void shouldRenderSelectWithLimitWithLockWrite() {
+
+		Table table = Table.create("foo");
+		LockMode lockMode = LockMode.PESSIMISTIC_WRITE;
+		Select select = StatementBuilder.select(table.asterisk()).from(table).limit(10).lock(lockMode).build();
+
+		String sql = SqlRenderer.create(factory.createRenderContext()).render(select);
+
+		assertThat(sql).isEqualTo("SELECT foo.* FROM foo LIMIT 10 FOR UPDATE");
+	}
+
+	@Test // DATAJDBC-498
+	public void shouldRenderSelectWithLimitWithLockRead() {
+
+		Table table = Table.create("foo");
+		LockMode lockMode = LockMode.PESSIMISTIC_READ;
+		Select select = StatementBuilder.select(table.asterisk()).from(table).limit(10).lock(lockMode).build();
+
+		String sql = SqlRenderer.create(factory.createRenderContext()).render(select);
+
+		assertThat(sql).isEqualTo("SELECT foo.* FROM foo LIMIT 10 LOCK IN SHARE MODE");
+	}
 }

spring-data-relational/src/test/java/org/springframework/data/relational/core/dialect/MySqlDialectUnitTests.java

@@ -15,15 +15,20 @@
  */
 package org.springframework.data.relational.core.dialect;
 
-import static org.assertj.core.api.Assertions.*;
-
 import org.junit.Test;
+import org.springframework.data.relational.core.sql.From;
+import org.springframework.data.relational.core.sql.LockMode;
+import org.springframework.data.relational.core.sql.LockOptions;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
 
 /**
  * Unit tests for {@link MySqlDialect}.
  *
  * @author Mark Paluch
  * @author Jens Schauder
+ * @author Myeonghyeon Lee
  */
 public class MySqlDialectUnitTests {
 
@@ -67,4 +72,15 @@ public class MySqlDialectUnitTests {
 
 		assertThat(abcQuoted).isEqualTo("`abc`");
 	}
+
+	@Test // DATAJDBC-498
+	public void shouldRenderLock() {
+
+		LockClause lock = MySqlDialect.INSTANCE.lock();
+		From from = mock(From.class);
+
+		assertThat(lock.getLock(new LockOptions(LockMode.PESSIMISTIC_WRITE, from))).isEqualTo("FOR UPDATE");
+		assertThat(lock.getLock(new LockOptions(LockMode.PESSIMISTIC_READ, from))).isEqualTo("LOCK IN SHARE MODE");
+		assertThat(lock.getClausePosition()).isEqualTo(LockClause.Position.AFTER_ORDER_BY);
+	}
 }

spring-data-relational/src/test/java/org/springframework/data/relational/core/dialect/PostgresDialectRenderingUnitTests.java

@@ -20,6 +20,7 @@ import static org.assertj.core.api.Assertions.*;
 import org.junit.Before;
 import org.junit.Test;
 
+import org.springframework.data.relational.core.sql.LockMode;
 import org.springframework.data.relational.core.sql.Select;
 import org.springframework.data.relational.core.sql.StatementBuilder;
 import org.springframework.data.relational.core.sql.Table;
@@ -31,6 +32,7 @@ import org.springframework.data.relational.core.sql.render.SqlRenderer;
  *
  * @author Mark Paluch
  * @author Jens Schauder
+ * @author Myeonghyeon Lee
  */
 public class PostgresDialectRenderingUnitTests {
 
@@ -97,4 +99,52 @@ public class PostgresDialectRenderingUnitTests {
 
 		assertThat(sql).isEqualTo("SELECT foo.* FROM foo LIMIT 10 OFFSET 20");
 	}
+
+	@Test // DATAJDBC-498
+	public void shouldRenderSelectWithLockWrite() {
+
+		Table table = Table.create("foo");
+		LockMode lockMode = LockMode.PESSIMISTIC_WRITE;
+		Select select = StatementBuilder.select(table.asterisk()).from(table).lock(lockMode).build();
+
+		String sql = SqlRenderer.create(factory.createRenderContext()).render(select);
+
+		assertThat(sql).isEqualTo("SELECT foo.* FROM foo FOR UPDATE OF foo");
+	}
+
+	@Test // DATAJDBC-498
+	public void shouldRenderSelectWithLockRead() {
+
+		Table table = Table.create("foo");
+		LockMode lockMode = LockMode.PESSIMISTIC_READ;
+		Select select = StatementBuilder.select(table.asterisk()).from(table).lock(lockMode).build();
+
+		String sql = SqlRenderer.create(factory.createRenderContext()).render(select);
+
+		assertThat(sql).isEqualTo("SELECT foo.* FROM foo FOR SHARE OF foo");
+	}
+
+	@Test // DATAJDBC-498
+	public void shouldRenderSelectWithLimitWithLockWrite() {
+
+		Table table = Table.create("foo");
+		LockMode lockMode = LockMode.PESSIMISTIC_WRITE;
+		Select select = StatementBuilder.select(table.asterisk()).from(table).limit(10).lock(lockMode).build();
+
+		String sql = SqlRenderer.create(factory.createRenderContext()).render(select);
+
+		assertThat(sql).isEqualTo("SELECT foo.* FROM foo LIMIT 10 FOR UPDATE OF foo");
+	}
+
+	@Test // DATAJDBC-498
+	public void shouldRenderSelectWithLimitWithLockRead() {
+
+		Table table = Table.create("foo");
+		LockMode lockMode = LockMode.PESSIMISTIC_READ;
+		Select select = StatementBuilder.select(table.asterisk()).from(table).limit(10).lock(lockMode).build();
+
+		String sql = SqlRenderer.create(factory.createRenderContext()).render(select);
+
+		assertThat(sql).isEqualTo("SELECT foo.* FROM foo LIMIT 10 FOR SHARE OF foo");
+	}
 }

spring-data-relational/src/test/java/org/springframework/data/relational/core/dialect/PostgresDialectUnitTests.java

@@ -17,13 +17,21 @@ package org.springframework.data.relational.core.dialect;
 
 import static org.assertj.core.api.Assertions.*;
 import static org.assertj.core.api.SoftAssertions.*;
+import static org.mockito.Mockito.*;
 
 import org.junit.Test;
+import org.springframework.data.relational.core.sql.From;
+import org.springframework.data.relational.core.sql.LockMode;
+import org.springframework.data.relational.core.sql.LockOptions;
+import org.springframework.data.relational.core.sql.Table;
+
+import java.util.Collections;
 
 /**
  * Unit tests for {@link PostgresDialect}.
  *
  * @author Mark Paluch
+ * @author Myeonghyeon Lee
  */
 public class PostgresDialectUnitTests {
 
@@ -71,4 +79,16 @@ public class PostgresDialectUnitTests {
 
 		assertThat(limit.getLimitOffset(20, 10)).isEqualTo("LIMIT 20 OFFSET 10");
 	}
+
+	@Test // DATAJDBC-498
+	public void shouldRenderLock() {
+
+		LockClause lock = PostgresDialect.INSTANCE.lock();
+		From from = mock(From.class);
+		when(from.getTables()).thenReturn(Collections.singletonList(Table.create("dummy_table")));
+
+		assertThat(lock.getLock(new LockOptions(LockMode.PESSIMISTIC_WRITE, from))).isEqualTo("FOR UPDATE OF dummy_table");
+		assertThat(lock.getLock(new LockOptions(LockMode.PESSIMISTIC_READ, from))).isEqualTo("FOR SHARE OF dummy_table");
+		assertThat(lock.getClausePosition()).isEqualTo(LockClause.Position.AFTER_ORDER_BY);
+	}
 }

spring-data-relational/src/test/java/org/springframework/data/relational/core/dialect/SqlServerDialectRenderingUnitTests.java

@@ -20,6 +20,7 @@ import static org.assertj.core.api.Assertions.*;
 import org.junit.Before;
 import org.junit.Test;
 
+import org.springframework.data.relational.core.sql.LockMode;
 import org.springframework.data.relational.core.sql.Select;
 import org.springframework.data.relational.core.sql.StatementBuilder;
 import org.springframework.data.relational.core.sql.Table;
@@ -31,6 +32,7 @@ import org.springframework.data.relational.core.sql.render.SqlRenderer;
  *
  * @author Mark Paluch
  * @author Jens Schauder
+ * @author Myeonghyeon Lee
  */
 public class SqlServerDialectRenderingUnitTests {
 
@@ -112,4 +114,82 @@ public class SqlServerDialectRenderingUnitTests {
 
 		assertThat(sql).isEqualTo("SELECT foo.* FROM foo ORDER BY column_1 OFFSET 20 ROWS FETCH NEXT 10 ROWS ONLY");
 	}
+
+	@Test // DATAJDBC-498
+	public void shouldRenderSelectWithLockWrite() {
+
+		Table table = Table.create("foo");
+		LockMode lockMode = LockMode.PESSIMISTIC_WRITE;
+		Select select = StatementBuilder.select(table.asterisk()).from(table).lock(lockMode).build();
+
+		String sql = SqlRenderer.create(factory.createRenderContext()).render(select);
+
+		assertThat(sql).isEqualTo(
+			"SELECT foo.* FROM foo WITH (UPDLOCK, ROWLOCK)");
+	}
+
+	@Test // DATAJDBC-498
+	public void shouldRenderSelectWithLockRead() {
+
+		Table table = Table.create("foo");
+		LockMode lockMode = LockMode.PESSIMISTIC_READ;
+		Select select = StatementBuilder.select(table.asterisk()).from(table).lock(lockMode).build();
+
+		String sql = SqlRenderer.create(factory.createRenderContext()).render(select);
+
+		assertThat(sql).isEqualTo(
+			"SELECT foo.* FROM foo WITH (HOLDLOCK, ROWLOCK)");
+	}
+
+	@Test // DATAJDBC-498
+	public void shouldRenderSelectWithLimitOffsetWithLockWrite() {
+
+		Table table = Table.create("foo");
+		LockMode lockMode = LockMode.PESSIMISTIC_WRITE;
+		Select select = StatementBuilder.select(table.asterisk()).from(table).limit(10).offset(20).lock(lockMode).build();
+
+		String sql = SqlRenderer.create(factory.createRenderContext()).render(select);
+
+		assertThat(sql).isEqualTo(
+			"SELECT foo.*, ROW_NUMBER() over (ORDER BY (SELECT 1)) AS __relational_row_number__ FROM foo WITH (UPDLOCK, ROWLOCK) ORDER BY __relational_row_number__ OFFSET 20 ROWS FETCH NEXT 10 ROWS ONLY");
+	}
+
+	@Test // DATAJDBC-498
+	public void shouldRenderSelectWithLimitOffsetWithLockRead() {
+
+		Table table = Table.create("foo");
+		LockMode lockMode = LockMode.PESSIMISTIC_READ;
+		Select select = StatementBuilder.select(table.asterisk()).from(table).limit(10).offset(20).lock(lockMode).build();
+
+		String sql = SqlRenderer.create(factory.createRenderContext()).render(select);
+
+		assertThat(sql).isEqualTo(
+			"SELECT foo.*, ROW_NUMBER() over (ORDER BY (SELECT 1)) AS __relational_row_number__ FROM foo WITH (HOLDLOCK, ROWLOCK) ORDER BY __relational_row_number__ OFFSET 20 ROWS FETCH NEXT 10 ROWS ONLY");
+	}
+
+	@Test // DATAJDBC-498
+	public void shouldRenderSelectWithLimitOffsetAndOrderByWithLockWrite() {
+
+		Table table = Table.create("foo");
+		LockMode lockMode = LockMode.PESSIMISTIC_WRITE;
+		Select select = StatementBuilder.select(table.asterisk()).from(table).orderBy(table.column("column_1")).limit(10)
+			.offset(20).lock(lockMode).build();
+
+		String sql = SqlRenderer.create(factory.createRenderContext()).render(select);
+
+		assertThat(sql).isEqualTo("SELECT foo.* FROM foo WITH (UPDLOCK, ROWLOCK) ORDER BY column_1 OFFSET 20 ROWS FETCH NEXT 10 ROWS ONLY");
+	}
+
+	@Test // DATAJDBC-498
+	public void shouldRenderSelectWithLimitOffsetAndOrderByWithLockRead() {
+
+		Table table = Table.create("foo");
+		LockMode lockMode = LockMode.PESSIMISTIC_READ;
+		Select select = StatementBuilder.select(table.asterisk()).from(table).orderBy(table.column("column_1")).limit(10)
+			.offset(20).lock(lockMode).build();
+
+		String sql = SqlRenderer.create(factory.createRenderContext()).render(select);
+
+		assertThat(sql).isEqualTo("SELECT foo.* FROM foo WITH (HOLDLOCK, ROWLOCK) ORDER BY column_1 OFFSET 20 ROWS FETCH NEXT 10 ROWS ONLY");
+	}
 }

spring-data-relational/src/test/java/org/springframework/data/relational/core/dialect/SqlServerDialectUnitTests.java

@@ -15,14 +15,20 @@
  */
 package org.springframework.data.relational.core.dialect;
 
-import static org.assertj.core.api.Assertions.*;
-
 import org.junit.Test;
+import org.springframework.data.relational.core.sql.From;
+import org.springframework.data.relational.core.sql.LockMode;
+import org.springframework.data.relational.core.sql.LockOptions;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
 
 /**
  * Unit tests for {@link SqlServerDialect}.
  *
  * @author Mark Paluch
+ * @author Myeonghyeon Lee
  */
 public class SqlServerDialectUnitTests {
 
@@ -59,4 +65,15 @@ public class SqlServerDialectUnitTests {
 
 		assertThat(limit.getLimitOffset(20, 10)).isEqualTo("OFFSET 10 ROWS FETCH NEXT 20 ROWS ONLY");
 	}
+
+	@Test // DATAJDBC-498
+	public void shouldRenderLock() {
+
+		LockClause lock = SqlServerDialect.INSTANCE.lock();
+		From from = mock(From.class);
+
+		assertThat(lock.getLock(new LockOptions(LockMode.PESSIMISTIC_WRITE, from))).isEqualTo("WITH (UPDLOCK, ROWLOCK)");
+		assertThat(lock.getLock(new LockOptions(LockMode.PESSIMISTIC_READ, from))).isEqualTo("WITH (HOLDLOCK, ROWLOCK)");
+		assertThat(lock.getClausePosition()).isEqualTo(LockClause.Position.AFTER_FROM_TABLE);
+	}
 }

spring-data-relational/src/test/java/org/springframework/data/relational/core/sql/SelectBuilderUnitTests.java

@@ -27,6 +27,7 @@ import org.springframework.data.relational.core.sql.Join.JoinType;
  * Unit tests for {@link SelectBuilder}.
  *
  * @author Mark Paluch
+ * @author Myeonghyeon Lee
  */
 public class SelectBuilderUnitTests {
 
@@ -147,4 +148,63 @@ public class SelectBuilderUnitTests {
 		assertThat(join.getType()).isEqualTo(JoinType.JOIN);
 	}
 
+	@Test // DATAJDBC-498
+	public void selectWithLock() {
+
+		SelectBuilder builder = StatementBuilder.select();
+
+		Table table = SQL.table("mytable");
+		Column foo = table.column("foo");
+		Column bar = table.column("bar");
+		LockMode lockMode = LockMode.PESSIMISTIC_WRITE;
+
+		Select select = builder.select(foo, bar).from(table).lock(lockMode).build();
+
+		CapturingVisitor visitor = new CapturingVisitor();
+		select.visit(visitor);
+
+		assertThat(visitor.enter).containsSequence(foo, table, bar, table, new From(table), table);
+		assertThat(select.getLockMode()).isEqualTo(lockMode);
+	}
+
+	@Test // DATAJDBC-498
+	public void selectWithWhereWithLock() {
+
+		SelectBuilder builder = StatementBuilder.select();
+
+		Table table = SQL.table("mytable");
+		Column foo = table.column("foo");
+
+		Comparison condition = foo.isEqualTo(SQL.literalOf("bar"));
+		LockMode lockMode = LockMode.PESSIMISTIC_WRITE;
+
+		Select select = builder.select(foo).from(table).where(condition).lock(lockMode).build();
+
+		CapturingVisitor visitor = new CapturingVisitor();
+		select.visit(visitor);
+
+		assertThat(visitor.enter).containsSequence(foo, table, new From(table), table, new Where(condition));
+		assertThat(select.getLockMode()).isEqualTo(lockMode);
+	}
+
+	@Test // DATAJDBC-498
+	public void orderByWithLock() {
+
+		SelectBuilder builder = StatementBuilder.select();
+
+		Table table = SQL.table("mytable");
+
+		Column foo = SQL.column("foo", table).as("foo");
+
+		OrderByField orderByField = OrderByField.from(foo).asc();
+		LockMode lockMode = LockMode.PESSIMISTIC_WRITE;
+
+		Select select = builder.select(foo).from(table).orderBy(orderByField).lock(lockMode).build();
+
+		CapturingVisitor visitor = new CapturingVisitor();
+		select.visit(visitor);
+
+		assertThat(visitor.enter).containsSequence(foo, table, new From(table), table, orderByField, foo);
+		assertThat(select.getLockMode()).isEqualTo(lockMode);
+	}
 }