DATACASS-606 - Support LWT for delete operations (DELETE .. IF EXISTS).

We now support lightweight transactions for entity deletion through DeleteOptions.

User user = …

DeleteOptions options = DeleteOptions.builder().withIfExists().build();
boolean applied = template.delete(user, options);

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraOperations.java

@@ -113,9 +113,8 @@ public interface AsyncCassandraOperations {
 	<T> ListenableFuture<List<T>> select(Statement statement, Class<T> entityClass) throws DataAccessException;
 
 	/**
-	 * Execute a {@code SELECT} query with paging and convert the result set to a {@link Slice} of entities.
-	 *
-	 * A sliced query translates the effective {@link Statement#getFetchSize() fetch size} to the page size.
+	 * Execute a {@code SELECT} query with paging and convert the result set to a {@link Slice} of entities. A sliced
+	 * query translates the effective {@link Statement#getFetchSize() fetch size} to the page size.
 	 *
 	 * @param statement the CQL statement, must not be {@literal null}.
 	 * @param entityClass The entity type must not be {@literal null}.
@@ -230,10 +229,9 @@ public interface AsyncCassandraOperations {
 	ListenableFuture<Long> count(Class<?> entityClass) throws DataAccessException;
 
 	/**
-	 * Returns the number of rows for the given entity class applying {@link Query}.
-	 *
-	 * This overridden method allows users to further refine the selection criteria using a {@link Query} predicate
-	 * to determine how many entities of the given {@link Class type} match the criteria.
+	 * Returns the number of rows for the given entity class applying {@link Query}. This overridden method allows users
+	 * to further refine the selection criteria using a {@link Query} predicate to determine how many entities of the
+	 * given {@link Class type} match the criteria.
 	 *
 	 * @param query user-provided count {@link Query} to execute; must not be {@literal null}.
 	 * @param entityClass {@link Class type} of the entity; must not be {@literal null}.
@@ -246,9 +244,9 @@ public interface AsyncCassandraOperations {
 	/**
 	 * Determine whether a row of {@code entityClass} with the given {@code id} exists.
 	 *
-	 * @param id Id value. For single primary keys it's the plain value. For composite primary keys either, it's
-	 * an instance of either {@link org.springframework.data.cassandra.core.mapping.PrimaryKeyClass}
-	 * or {@link org.springframework.data.cassandra.core.mapping.MapId}. Must not be {@literal null}.
+	 * @param id Id value. For single primary keys it's the plain value. For composite primary keys either, it's an
+	 *          instance of either {@link org.springframework.data.cassandra.core.mapping.PrimaryKeyClass} or
+	 *          {@link org.springframework.data.cassandra.core.mapping.MapId}. Must not be {@literal null}.
 	 * @param entityClass {@link Class type} of the entity; must not be {@literal null}.
 	 * @return {@literal true} if the object exists.
 	 * @throws DataAccessException if any problem occurs while executing the query.
@@ -338,6 +336,20 @@ public interface AsyncCassandraOperations {
 	 */
 	ListenableFuture<WriteResult> delete(Object entity, QueryOptions options) throws DataAccessException;
 
+	/**
+	 * Delete the given entity applying {@link DeleteOptions} and return the entity if the delete was applied.
+	 *
+	 * @param entity must not be {@literal null}.
+	 * @param options must not be {@literal null}.
+	 * @return the {@link WriteResult} for this operation.
+	 * @throws DataAccessException if there is any problem executing the query.
+	 * @see DeleteOptions#empty()
+	 * @since 2.2
+	 */
+	default ListenableFuture<WriteResult> delete(Object entity, DeleteOptions options) throws DataAccessException {
+		return delete(entity, (QueryOptions) options);
+	}
+
 	/**
 	 * Remove the given object from the table by id.
 	 *

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/AsyncCassandraTemplate.java

@@ -249,7 +249,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
 		Assert.notNull(statement, "Statement must not be null");
 		Assert.notNull(entityClass, "Entity type must not be null");
 
-		Function<Row, T> mapper = getMapper(entityClass, entityClass, QueryUtils.getTableName(statement));
+		Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
 
 		return getAsyncCqlOperations().query(statement, (row, rowNum) -> mapper.apply(row));
 	}
@@ -265,10 +265,10 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
 
 		ListenableFuture<ResultSet> resultSet = getAsyncCqlOperations().queryForResultSet(statement);
 
-		Function<Row, T> mapper = getMapper(entityClass, entityClass, QueryUtils.getTableName(statement));
+		Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
 
 		return new MappingListenableFutureAdapter<>(resultSet,
-				rs -> QueryUtils.readSlice(rs, (row, rowNum) -> mapper.apply(row), 0, getEffectiveFetchSize(statement)));
+				rs -> EntityQueryUtils.readSlice(rs, (row, rowNum) -> mapper.apply(row), 0, getEffectiveFetchSize(statement)));
 	}
 
 	/* (non-Javadoc)
@@ -282,7 +282,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
 		Assert.notNull(entityConsumer, "Entity Consumer must not be empty");
 		Assert.notNull(entityClass, "Entity type must not be null");
 
-		Function<Row, T> mapper = getMapper(entityClass, entityClass, QueryUtils.getTableName(statement));
+		Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
 
 		return getAsyncCqlOperations().query(statement, row -> {
 			entityConsumer.accept(mapper.apply(row));
@@ -504,7 +504,8 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
 
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
 		CqlIdentifier tableName = persistentEntity.getTableName();
-		Insert insert = QueryUtils.createInsertQuery(tableName.toCql(), entity, options, getConverter(), persistentEntity);
+		Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entity, options, getConverter(),
+				persistentEntity);
 
 		maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, insert));
 
@@ -533,7 +534,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
 		Assert.notNull(options, "UpdateOptions must not be null");
 
 		CqlIdentifier tableName = getTableName(entity);
-		Update update = QueryUtils.createUpdateQuery(tableName.toCql(), entity, options, getConverter());
+		Update update = EntityQueryUtils.createUpdateQuery(tableName.toCql(), entity, options, getConverter());
 
 		maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, update));
 
@@ -562,7 +563,7 @@ public class AsyncCassandraTemplate implements AsyncCassandraOperations, Applica
 		Assert.notNull(options, "QueryOptions must not be null");
 
 		CqlIdentifier tableName = getTableName(entity);
-		Delete delete = QueryUtils.createDeleteQuery(tableName.toCql(), entity, options, getConverter());
+		Delete delete = EntityQueryUtils.createDeleteQuery(tableName.toCql(), entity, options, getConverter());
 
 		maybeEmitEvent(new BeforeDeleteEvent<>(delete, entity.getClass(), tableName));
 

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraOperations.java

@@ -134,9 +134,8 @@ public interface CassandraOperations extends FluentCassandraOperations {
 	<T> List<T> select(Statement statement, Class<T> entityClass) throws DataAccessException;
 
 	/**
-	 * Execute a {@code SELECT} query with paging and convert the result set to a {@link Slice} of entities.
-	 *
-	 * A sliced query translates the effective {@link Statement#getFetchSize() fetch size} to the page size.
+	 * Execute a {@code SELECT} query with paging and convert the result set to a {@link Slice} of entities. A sliced
+	 * query translates the effective {@link Statement#getFetchSize() fetch size} to the page size.
 	 *
 	 * @param statement the CQL statement, must not be {@literal null}.
 	 * @param entityClass The entity type must not be {@literal null}.
@@ -257,10 +256,9 @@ public interface CassandraOperations extends FluentCassandraOperations {
 	long count(Class<?> entityClass) throws DataAccessException;
 
 	/**
-	 * Returns the number of rows for the given entity class applying {@link Query}.
-	 *
-	 * This overridden method allows users to further refine the selection criteria using a {@link Query} predicate
-	 * to determine how many entities of the given {@link Class type} match the criteria.
+	 * Returns the number of rows for the given entity class applying {@link Query}. This overridden method allows users
+	 * to further refine the selection criteria using a {@link Query} predicate to determine how many entities of the
+	 * given {@link Class type} match the criteria.
 	 *
 	 * @param query user-defined count {@link Query} to execute; must not be {@literal null}.
 	 * @param entityClass {@link Class type} of the entity; must not be {@literal null}.
@@ -273,9 +271,9 @@ public interface CassandraOperations extends FluentCassandraOperations {
 	/**
 	 * Determine whether a row of {@code entityClass} with the given {@code id} exists.
 	 *
-	 * @param id Id value. For single primary keys it's the plain value. For composite primary keys either, it's
-	 * an instance of either {@link org.springframework.data.cassandra.core.mapping.PrimaryKeyClass}
-	 * or {@link org.springframework.data.cassandra.core.mapping.MapId}. Must not be {@literal null}.
+	 * @param id Id value. For single primary keys it's the plain value. For composite primary keys either, it's an
+	 *          instance of either {@link org.springframework.data.cassandra.core.mapping.PrimaryKeyClass} or
+	 *          {@link org.springframework.data.cassandra.core.mapping.MapId}. Must not be {@literal null}.
 	 * @param entityClass {@link Class type} of the entity; must not be {@literal null}.
 	 * @return {@literal true} if the object exists.
 	 * @throws DataAccessException if any problem occurs while executing the query.
@@ -365,6 +363,20 @@ public interface CassandraOperations extends FluentCassandraOperations {
 	 */
 	WriteResult delete(Object entity, QueryOptions options) throws DataAccessException;
 
+	/**
+	 * Delete the given entity applying {@link DeleteOptions} and return the entity if the delete was applied.
+	 *
+	 * @param entity must not be {@literal null}.
+	 * @param options must not be {@literal null}.
+	 * @return the {@link WriteResult} for this operation.
+	 * @throws DataAccessException if there is any problem executing the query.
+	 * @see DeleteOptions#empty()
+	 * @since 2.2
+	 */
+	default WriteResult delete(Object entity, DeleteOptions options) throws DataAccessException {
+		return delete(entity, (QueryOptions) options);
+	}
+
 	/**
 	 * Remove the given object from the table by id.
 	 *

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraTemplate.java

@@ -258,7 +258,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		Assert.notNull(statement, "Statement must not be null");
 		Assert.notNull(entityClass, "Entity type must not be null");
 
-		Function<Row, T> mapper = getMapper(entityClass, entityClass, QueryUtils.getTableName(statement));
+		Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
 
 		return getCqlOperations().query(statement, (row, rowNum) -> mapper.apply(row));
 	}
@@ -274,9 +274,10 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 
 		ResultSet resultSet = getCqlOperations().queryForResultSet(statement);
 
-		Function<Row, T> mapper = getMapper(entityClass, entityClass, QueryUtils.getTableName(statement));
+		Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement));
 
-		return QueryUtils.readSlice(resultSet, (row, rowNum) -> mapper.apply(row), 0, getEffectiveFetchSize(statement));
+		return EntityQueryUtils.readSlice(resultSet, (row, rowNum) -> mapper.apply(row), 0,
+				getEffectiveFetchSize(statement));
 	}
 
 	/* (non-Javadoc)
@@ -291,7 +292,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		ResultSet resultSet = getCqlOperations().queryForResultSet(statement);
 
 		return StreamSupport.stream(resultSet.spliterator(), false)
-				.map(getMapper(entityClass, entityClass, QueryUtils.getTableName(statement)));
+				.map(getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(statement)));
 	}
 
 	/* (non-Javadoc)
@@ -551,7 +552,8 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
 
-		Insert insert = QueryUtils.createInsertQuery(tableName.toCql(), entity, options, getConverter(), persistentEntity);
+		Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entity, options, getConverter(),
+				persistentEntity);
 
 		maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, insert));
 
@@ -581,7 +583,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		Assert.notNull(options, "UpdateOptions must not be null");
 
 		CqlIdentifier tableName = getTableName(entity);
-		Update update = QueryUtils.createUpdateQuery(tableName.toCql(), entity, options, getConverter());
+		Update update = EntityQueryUtils.createUpdateQuery(tableName.toCql(), entity, options, getConverter());
 
 		maybeEmitEvent(new BeforeSaveEvent<>(entity, tableName, update));
 
@@ -611,7 +613,7 @@ public class CassandraTemplate implements CassandraOperations, ApplicationEventP
 		Assert.notNull(options, "QueryOptions must not be null");
 
 		CqlIdentifier tableName = getTableName(entity);
-		Delete delete = QueryUtils.createDeleteQuery(tableName.toCql(), entity, options, getConverter());
+		Delete delete = EntityQueryUtils.createDeleteQuery(tableName.toCql(), entity, options, getConverter());
 
 		maybeEmitEvent(new BeforeDeleteEvent<>(delete, entity.getClass(), tableName));
 

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/DeleteOptions.java

@@ -0,0 +1,257 @@
+/*
+ * Copyright 2019 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
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.cassandra.core;
+
+import lombok.EqualsAndHashCode;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.data.cassandra.core.cql.WriteOptions;
+import org.springframework.lang.Nullable;
+
+import com.datastax.driver.core.ConsistencyLevel;
+import com.datastax.driver.core.policies.RetryPolicy;
+
+/**
+ * Extension to {@link WriteOptions} for use with {@code DELETE} operations.
+ *
+ * @author Mark Paluch
+ * @since 2.2
+ */
+@EqualsAndHashCode(callSuper = true)
+public class DeleteOptions extends WriteOptions {
+
+	private static final DeleteOptions EMPTY = new DeleteOptionsBuilder().build();
+
+	private boolean ifExists;
+
+	private DeleteOptions(@Nullable ConsistencyLevel consistencyLevel, @Nullable RetryPolicy retryPolicy,
+			@Nullable Boolean tracing, @Nullable Integer fetchSize, Duration readTimeout, Duration ttl,
+			@Nullable Long timestamp, boolean ifExists) {
+
+		super(consistencyLevel, retryPolicy, tracing, fetchSize, readTimeout, ttl, timestamp);
+
+		this.ifExists = ifExists;
+	}
+
+	/**
+	 * Create a new {@link DeleteOptionsBuilder}.
+	 *
+	 * @return a new {@link DeleteOptionsBuilder}.
+	 */
+	public static DeleteOptionsBuilder builder() {
+		return new DeleteOptionsBuilder();
+	}
+
+	/**
+	 * Create default {@link DeleteOptions}.
+	 *
+	 * @return default {@link DeleteOptions}.
+	 */
+	public static DeleteOptions empty() {
+		return EMPTY;
+	}
+
+	/**
+	 * Create a new {@link DeleteOptionsBuilder} to mutate properties of this {@link DeleteOptions}.
+	 *
+	 * @return a new {@link DeleteOptionsBuilder} initialized with this {@link DeleteOptions}.
+	 */
+	@Override
+	public DeleteOptionsBuilder mutate() {
+		return new DeleteOptionsBuilder(this);
+	}
+
+	/**
+	 * @return {@literal true} to apply {@code IF EXISTS} to {@code UPDATE} operations.
+	 */
+	public boolean isIfExists() {
+		return this.ifExists;
+	}
+
+	/**
+	 * Builder for {@link DeleteOptions}.
+	 *
+	 * @author Mark Paluch
+	 */
+	public static class DeleteOptionsBuilder extends WriteOptionsBuilder {
+
+		private boolean ifExists;
+
+		private DeleteOptionsBuilder() {}
+
+		private DeleteOptionsBuilder(DeleteOptions deleteOptions) {
+
+			super(deleteOptions);
+
+			this.ifExists = deleteOptions.ifExists;
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#consistencyLevel(com.datastax.driver.core.ConsistencyLevel)
+		 */
+		@Override
+		public DeleteOptionsBuilder consistencyLevel(ConsistencyLevel consistencyLevel) {
+
+			super.consistencyLevel(consistencyLevel);
+			return this;
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#retryPolicy(com.datastax.driver.core.policies.RetryPolicy)
+		 */
+		@Override
+		public DeleteOptionsBuilder retryPolicy(RetryPolicy driverRetryPolicy) {
+
+			super.retryPolicy(driverRetryPolicy);
+			return this;
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#fetchSize(int)
+		 */
+		@Override
+		public DeleteOptionsBuilder fetchSize(int fetchSize) {
+
+			super.fetchSize(fetchSize);
+			return this;
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#readTimeout(long)
+		 */
+		@Override
+		public DeleteOptionsBuilder readTimeout(long readTimeout) {
+
+			super.readTimeout(readTimeout);
+			return this;
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#readTimeout(long, java.util.concurrent.TimeUnit)
+		 */
+		@Override
+		@Deprecated
+		public DeleteOptionsBuilder readTimeout(long readTimeout, TimeUnit timeUnit) {
+
+			super.readTimeout(readTimeout, timeUnit);
+			return this;
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#readTimeout(java.time.Duration)
+		 */
+		@Override
+		public DeleteOptionsBuilder readTimeout(Duration readTimeout) {
+
+			super.readTimeout(readTimeout);
+			return this;
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#ttl(java.time.Duration)
+		 */
+		@Override
+		public DeleteOptionsBuilder ttl(Duration ttl) {
+
+			super.ttl(ttl);
+			return this;
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#tracing(boolean)
+		 */
+		@Override
+		public DeleteOptionsBuilder tracing(boolean tracing) {
+
+			super.tracing(tracing);
+			return this;
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#withTracing()
+		 */
+		@Override
+		public DeleteOptionsBuilder withTracing() {
+
+			super.withTracing();
+			return this;
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#ttl(int)
+		 */
+		public DeleteOptionsBuilder ttl(int ttl) {
+
+			super.ttl(ttl);
+			return this;
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#timestamp(long)
+		 */
+		@Override
+		public DeleteOptionsBuilder timestamp(long timestamp) {
+
+			super.timestamp(timestamp);
+			return this;
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.cassandra.core.cql.WriteOptions.WriteOptionsBuilder#timestamp(java.time.Instant)
+		 */
+		@Override
+		public DeleteOptionsBuilder timestamp(Instant timestamp) {
+
+			super.timestamp(timestamp);
+			return this;
+		}
+
+		/**
+		 * Use light-weight transactions by applying {@code IF EXISTS}.
+		 *
+		 * @return {@code this} {@link DeleteOptionsBuilder}
+		 */
+		public DeleteOptionsBuilder withIfExists() {
+			return ifExists(true);
+		}
+
+		/**
+		 * Use light-weight transactions by applying {@code IF EXISTS}.
+		 *
+		 * @param ifNotExists {@literal true} to enable {@code IF EXISTS}.
+		 * @return {@code this} {@link DeleteOptionsBuilder}
+		 */
+		public DeleteOptionsBuilder ifExists(boolean ifNotExists) {
+
+			this.ifExists = ifNotExists;
+
+			return this;
+		}
+
+		/**
+		 * Builds a new {@link DeleteOptions} with the configured values.
+		 *
+		 * @return a new {@link DeleteOptions} with the configured values
+		 */
+		public DeleteOptions build() {
+			return new DeleteOptions(this.consistencyLevel, this.retryPolicy, this.tracing, this.fetchSize, this.readTimeout,
+					this.ttl, this.timestamp, this.ifExists);
+		}
+	}
+}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/EntityQueryUtils.java

@@ -61,7 +61,7 @@ import com.google.common.collect.Iterators;
  * @author Mark Paluch
  * @since 2.0
  */
-class QueryUtils {
+class EntityQueryUtils {
 
 	private static final Pattern FROM_REGEX = Pattern.compile(" FROM ([\"]?[\\w]*[\\\\.]?[\\w]*[\"]?)[\\s]?",
 			Pattern.CASE_INSENSITIVE);
@@ -155,7 +155,7 @@ class QueryUtils {
 		Delete delete = deleteSelection.from(tableName);
 
 		if (options instanceof WriteOptions) {
-			QueryOptionsUtil.addWriteOptions(delete, (WriteOptions) options);
+			addWriteOptions(delete, (WriteOptions) options);
 		} else {
 			QueryOptionsUtil.addQueryOptions(delete, options);
 		}
@@ -322,7 +322,16 @@ class QueryUtils {
 
 		Assert.notNull(delete, "Delete must not be null");
 
-		QueryOptionsUtil.addQueryOptions(delete, writeOptions);
+		QueryOptionsUtil.addWriteOptions(delete, writeOptions);
+
+		if (writeOptions instanceof DeleteOptions) {
+
+			DeleteOptions deleteOptions = (DeleteOptions) writeOptions;
+
+			if (deleteOptions.isIfExists()) {
+				delete.where().ifExists();
+			}
+		}
 
 		return delete;
 	}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraOperations.java

@@ -203,10 +203,9 @@ public interface ReactiveCassandraOperations extends ReactiveFluentCassandraOper
 	Mono<Long> count(Class<?> entityClass) throws DataAccessException;
 
 	/**
-	 * Returns the number of rows for the given entity class applying {@link Query}.
-	 *
-	 * This overridden method allows users to further refine the selection criteria using a {@link Query} predicate
-	 * to determine how many entities of the given {@link Class type} match the criteria.
+	 * Returns the number of rows for the given entity class applying {@link Query}. This overridden method allows users
+	 * to further refine the selection criteria using a {@link Query} predicate to determine how many entities of the
+	 * given {@link Class type} match the criteria.
 	 *
 	 * @param query user-defined count {@link Query} to execute; must not be {@literal null}.
 	 * @param entityClass {@link Class type} of the entity; must not be {@literal null}.
@@ -219,9 +218,9 @@ public interface ReactiveCassandraOperations extends ReactiveFluentCassandraOper
 	/**
 	 * Determine whether a row of {@code entityClass} with the given {@code id} exists.
 	 *
-	 * @param id Id value. For single primary keys it's the plain value. For composite primary keys either, it's
-	 * an instance of either {@link org.springframework.data.cassandra.core.mapping.PrimaryKeyClass}
-	 * or {@link org.springframework.data.cassandra.core.mapping.MapId}. Must not be {@literal null}.
+	 * @param id Id value. For single primary keys it's the plain value. For composite primary keys either, it's an
+	 *          instance of either {@link org.springframework.data.cassandra.core.mapping.PrimaryKeyClass} or
+	 *          {@link org.springframework.data.cassandra.core.mapping.MapId}. Must not be {@literal null}.
 	 * @param entityClass {@link Class type} of the entity; must not be {@literal null}.
 	 * @return {@literal true} if the object exists.
 	 * @throws DataAccessException if any problem occurs while executing the query.
@@ -311,6 +310,20 @@ public interface ReactiveCassandraOperations extends ReactiveFluentCassandraOper
 	 */
 	Mono<WriteResult> delete(Object entity, QueryOptions options) throws DataAccessException;
 
+	/**
+	 * Delete the given entity applying {@link QueryOptions} and emit the entity if the delete was applied.
+	 *
+	 * @param entity must not be {@literal null}.
+	 * @param options must not be {@literal null}.
+	 * @return the {@link WriteResult} for this operation.
+	 * @throws DataAccessException if there is any problem issuing the execution.
+	 * @see DeleteOptions#empty()
+	 * @since 2.2
+	 */
+	default Mono<WriteResult> delete(Object entity, DeleteOptions options) throws DataAccessException {
+		return delete(entity, (QueryOptions) options);
+	}
+
 	/**
 	 * Remove the given object from the table by id.
 	 *

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplate.java

@@ -239,7 +239,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
 		Assert.notNull(cql, "Statement must not be null");
 		Assert.notNull(entityClass, "Entity type must not be null");
 
-		Function<Row, T> mapper = getMapper(entityClass, entityClass, QueryUtils.getTableName(cql));
+		Function<Row, T> mapper = getMapper(entityClass, entityClass, EntityQueryUtils.getTableName(cql));
 
 		return getReactiveCqlOperations().query(cql, (row, rowNum) -> mapper.apply(row));
 	}
@@ -263,7 +263,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
 			Integer effectiveFetchSize = tuple.getT2();
 
 			return resultSet.availableRows().collectList().map(it -> {
-				return QueryUtils.readSlice(it, resultSet.getExecutionInfo().getPagingState(), rowMapper, 1,
+				return EntityQueryUtils.readSlice(it, resultSet.getExecutionInfo().getPagingState(), rowMapper, 1,
 						effectiveFetchSize);
 			});
 
@@ -488,7 +488,8 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
 
 		CassandraPersistentEntity<?> persistentEntity = getRequiredPersistentEntity(entity.getClass());
 
-		Insert insert = QueryUtils.createInsertQuery(tableName.toCql(), entity, options, getConverter(), persistentEntity);
+		Insert insert = EntityQueryUtils.createInsertQuery(tableName.toCql(), entity, options, getConverter(),
+				persistentEntity);
 
 		// noinspection ConstantConditions
 		Mono<EntityWriteResult<T>> result = getReactiveCqlOperations() //
@@ -518,7 +519,7 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
 		Assert.notNull(options, "UpdateOptions must not be null");
 
 		CqlIdentifier tableName = getTableName(entity);
-		Update update = QueryUtils.createUpdateQuery(tableName.toCql(), entity, options, getConverter());
+		Update update = EntityQueryUtils.createUpdateQuery(tableName.toCql(), entity, options, getConverter());
 
 		Mono<EntityWriteResult<T>> result = getReactiveCqlOperations() //
 				.execute(new StatementCallback(update)) //
@@ -547,7 +548,8 @@ public class ReactiveCassandraTemplate implements ReactiveCassandraOperations, A
 		Assert.notNull(options, "QueryOptions must not be null");
 
 		CqlIdentifier tableName = getTableName(entity);
-		Delete delete = QueryUtils.createDeleteQuery(tableName.toCql(), entity, options, getConverter());
+
+		Delete delete = EntityQueryUtils.createDeleteQuery(tableName.toCql(), entity, options, getConverter());
 
 		Mono<WriteResult> result = getReactiveCqlOperations() //
 				.execute(new StatementCallback(delete)) //

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/StatementFactory.java

@@ -313,7 +313,7 @@ public class StatementFactory {
 
 		query.getQueryOptions().ifPresent(queryOptions -> {
 			if (queryOptions instanceof WriteOptions) {
-				QueryUtils.addWriteOptions(update, (WriteOptions) queryOptions);
+				EntityQueryUtils.addWriteOptions(update, (WriteOptions) queryOptions);
 			} else {
 				QueryOptionsUtil.addQueryOptions(update, queryOptions);
 			}
@@ -454,7 +454,7 @@ public class StatementFactory {
 
 		query.getQueryOptions().ifPresent(queryOptions -> {
 			if (queryOptions instanceof WriteOptions) {
-				QueryUtils.addWriteOptions(delete, (WriteOptions) queryOptions);
+				EntityQueryUtils.addWriteOptions(delete, (WriteOptions) queryOptions);
 			} else {
 				QueryOptionsUtil.addQueryOptions(delete, queryOptions);
 			}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/cql/QueryOptionsUtil.java

@@ -126,7 +126,7 @@ public abstract class QueryOptionsUtil {
 	 */
 	public static Delete addWriteOptions(Delete delete, WriteOptions writeOptions) {
 
-		Assert.notNull(delete, "Update must not be null");
+		Assert.notNull(delete, "Delete must not be null");
 
 		addQueryOptions(delete, writeOptions);
 

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/AsyncCassandraTemplateIntegrationTests.java

@@ -284,6 +284,31 @@ public class AsyncCassandraTemplateIntegrationTests extends AbstractKeyspaceCrea
 		assertThat(getUser(user.getId())).isNull();
 	}
 
+	@Test // DATACASS-606
+	public void deleteShouldRemoveEntityWithLwt() {
+
+		DeleteOptions lwtOptions = DeleteOptions.builder().withIfExists().build();
+
+		User user = new User("heisenberg", "Walter", "White");
+		getUninterruptibly(template.insert(user));
+
+		assertThat(getUninterruptibly(template.delete(user, lwtOptions)).wasApplied()).isTrue();
+		assertThat(getUninterruptibly(template.delete(user, lwtOptions)).wasApplied()).isFalse();
+	}
+
+	@Test // DATACASS-606
+	public void deleteByQueryShouldRemoveEntityWithLwt() {
+
+		DeleteOptions lwtOptions = DeleteOptions.builder().withIfExists().build();
+
+		User user = new User("heisenberg", "Walter", "White");
+		getUninterruptibly(template.insert(user));
+
+		Query query = Query.query(where("id").is("heisenberg")).queryOptions(lwtOptions);
+		assertThat(getUninterruptibly(template.delete(query, User.class))).isTrue();
+		assertThat(getUninterruptibly(template.delete(query, User.class))).isFalse();
+	}
+
 	@Test // DATACASS-56
 	public void shouldPageRequests() {
 

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/CassandraTemplateIntegrationTests.java

@@ -348,6 +348,31 @@ public class CassandraTemplateIntegrationTests extends AbstractKeyspaceCreatingI
 		assertThat(template.selectOneById(user.getId(), User.class)).isNull();
 	}
 
+	@Test // DATACASS-606
+	public void deleteShouldRemoveEntityWithLwt() {
+
+		DeleteOptions lwtOptions = DeleteOptions.builder().withIfExists().build();
+
+		User user = new User("heisenberg", "Walter", "White");
+		template.insert(user);
+
+		assertThat(template.delete(user, lwtOptions).wasApplied()).isTrue();
+		assertThat(template.delete(user, lwtOptions).wasApplied()).isFalse();
+	}
+
+	@Test // DATACASS-606
+	public void deleteByQueryShouldRemoveEntityWithLwt() {
+
+		DeleteOptions lwtOptions = DeleteOptions.builder().withIfExists().build();
+
+		User user = new User("heisenberg", "Walter", "White");
+		template.insert(user);
+
+		Query query = Query.query(where("id").is("heisenberg")).queryOptions(lwtOptions);
+		assertThat(template.delete(query, User.class)).isTrue();
+		assertThat(template.delete(query, User.class)).isFalse();
+	}
+
 	@Test // DATACASS-182
 	public void stream() {
 

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/EntityQueryUtilsUnitTests.java

@@ -23,16 +23,17 @@ import org.springframework.data.cassandra.core.cql.CqlIdentifier;
 import org.springframework.data.cassandra.domain.User;
 
 import com.datastax.driver.core.SimpleStatement;
+import com.datastax.driver.core.querybuilder.Delete;
 import com.datastax.driver.core.querybuilder.Insert;
 import com.datastax.driver.core.querybuilder.QueryBuilder;
 import com.datastax.driver.core.querybuilder.Select;
 
 /**
- * Unit tests for {@link QueryUtils}.
+ * Unit tests for {@link EntityQueryUtils}.
  *
  * @author Mark Paluch
  */
-public class QueryUtilsUnitTests {
+public class EntityQueryUtilsUnitTests {
 
 	private final MappingCassandraConverter converter = new MappingCassandraConverter();
 
@@ -41,7 +42,7 @@ public class QueryUtilsUnitTests {
 
 		Select select = QueryBuilder.select().from("keyspace", "table");
 
-		CqlIdentifier tableName = QueryUtils.getTableName(select);
+		CqlIdentifier tableName = EntityQueryUtils.getTableName(select);
 
 		assertThat(tableName).isEqualTo(CqlIdentifier.of("table"));
 	}
@@ -49,16 +50,16 @@ public class QueryUtilsUnitTests {
 	@Test // DATACASS-106
 	public void shouldRetrieveTableNameFromSimpleStatement() {
 
-		assertThat(QueryUtils.getTableName(new SimpleStatement("SELECT * FROM table")))
+		assertThat(EntityQueryUtils.getTableName(new SimpleStatement("SELECT * FROM table")))
 				.isEqualTo(CqlIdentifier.of("table"));
-		assertThat(QueryUtils.getTableName(new SimpleStatement("SELECT * FROM foo.table where")))
+		assertThat(EntityQueryUtils.getTableName(new SimpleStatement("SELECT * FROM foo.table where")))
 				.isEqualTo(CqlIdentifier.of("table"));
 	}
 
 	@Test // DATACASS-106
 	public void shouldRetrieveQuotedTableNameFromSimpleStatement() {
 
-		CqlIdentifier tableName = QueryUtils.getTableName(new SimpleStatement("SELECT * from \"table\""));
+		CqlIdentifier tableName = EntityQueryUtils.getTableName(new SimpleStatement("SELECT * from \"table\""));
 
 		assertThat(tableName).isEqualTo(CqlIdentifier.of("table"));
 	}
@@ -67,10 +68,22 @@ public class QueryUtilsUnitTests {
 	public void shouldCreateInsertQuery() {
 
 		User user = new User("heisenberg", "Walter", "White");
-		Insert insert = QueryUtils.createInsertQuery("user", user, InsertOptions.builder().withIfNotExists().build(),
+		Insert insert = EntityQueryUtils.createInsertQuery("user", user, InsertOptions.builder().withIfNotExists().build(),
 				converter, converter.getMappingContext().getRequiredPersistentEntity(User.class));
 
 		assertThat(insert.toString())
 				.isEqualTo("INSERT INTO user (firstname,id,lastname) VALUES ('Walter','heisenberg','White') IF NOT EXISTS;");
 	}
+
+	@Test // DATACASS-606
+	public void shouldConsiderDeleteIfExists() {
+
+		User user = new User("heisenberg", "Walter", "White");
+
+		DeleteOptions options = DeleteOptions.builder().withIfExists().build();
+
+		Delete delete = EntityQueryUtils.createDeleteQuery("foo", user, options, converter);
+
+		assertThat(delete.toString()).isEqualTo("DELETE FROM foo WHERE id='heisenberg' IF EXISTS;");
+	}
 }

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/core/ReactiveCassandraTemplateIntegrationTests.java

@@ -276,6 +276,46 @@ public class ReactiveCassandraTemplateIntegrationTests extends AbstractKeyspaceC
 		verifyUser(user.getId()).verifyComplete();
 	}
 
+	@Test // DATACASS-606
+	public void deleteShouldRemoveEntityWithLwt() {
+
+		DeleteOptions lwtOptions = DeleteOptions.builder().withIfExists().build();
+
+		User user = new User("heisenberg", "Walter", "White");
+		StepVerifier.create(template.insert(user)).expectNextCount(1).verifyComplete();
+
+		template.delete(user, lwtOptions).map(WriteResult::wasApplied) //
+				.as(StepVerifier::create) //
+				.expectNext(true) //
+				.verifyComplete();
+
+		template.delete(user, lwtOptions).map(WriteResult::wasApplied) //
+				.as(StepVerifier::create) //
+				.expectNext(false) //
+				.verifyComplete();
+	}
+
+	@Test // DATACASS-606
+	public void deleteByQueryShouldRemoveEntityWithLwt() {
+
+		DeleteOptions lwtOptions = DeleteOptions.builder().withIfExists().build();
+
+		User user = new User("heisenberg", "Walter", "White");
+		StepVerifier.create(template.insert(user)).expectNextCount(1).verifyComplete();
+
+		Query query = Query.query(where("id").is("heisenberg")).queryOptions(lwtOptions);
+
+		template.delete(query, User.class) //
+				.as(StepVerifier::create) //
+				.expectNext(true) //
+				.verifyComplete();
+
+		template.delete(query, User.class) //
+				.as(StepVerifier::create) //
+				.expectNext(false) //
+				.verifyComplete();
+	}
+
 	@Test // DATACASS-343
 	public void shouldSelectByQueryWithSorting() {