Add overloads to CassandraBatchOperations.insert(…)/update(…), and delete(…) accepting WriteOptions.

CassandraBatchOperations and its reactive variant provide now overloads accepting a single entity and WriteOptions. We also guard methods accepting varargs against being called with an accidental WriteOptions argument by inspecting the array of objects.

Closes #1135

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

@@ -15,7 +15,10 @@
  */
 package org.springframework.data.cassandra.core;
 
+import java.util.Collections;
+
 import org.springframework.data.cassandra.core.cql.WriteOptions;
+import org.springframework.util.Assert;
 
 /**
  * Batch operations for insert/update/delete actions on a table. {@link CassandraBatchOperations} use logged Cassandra
@@ -54,6 +57,22 @@ public interface CassandraBatchOperations {
 	 */
 	CassandraBatchOperations withTimestamp(long timestamp);
 
+	/**
+	 * Add an insert to the batch.
+	 *
+	 * @param entity the entity to insert; must not be {@literal null}.
+	 * @param options the WriteOptions to apply; must not be {@literal null}.
+	 * @return {@code this} {@link CassandraBatchOperations}.
+	 * @throws IllegalStateException if the batch was already executed.
+	 * @since 3.2.2
+	 */
+	default CassandraBatchOperations insert(Object entity, WriteOptions options) {
+
+		Assert.notNull(entity, "Entity must not be null");
+
+		return insert(Collections.singleton(entity), options);
+	}
+
 	/**
 	 * Add an array of inserts to the batch.
 	 *
@@ -84,6 +103,22 @@ public interface CassandraBatchOperations {
 	 */
 	CassandraBatchOperations insert(Iterable<?> entities, WriteOptions options);
 
+	/**
+	 * Add an update to the batch.
+	 *
+	 * @param entity the entity to update; must not be {@literal null}.
+	 * @param options the WriteOptions to apply; must not be {@literal null}.
+	 * @return {@code this} {@link CassandraBatchOperations}.
+	 * @throws IllegalStateException if the batch was already executed.
+	 * @since 3.2.2
+	 */
+	default CassandraBatchOperations update(Object entity, WriteOptions options) {
+
+		Assert.notNull(entity, "Entity must not be null");
+
+		return insert(Collections.singleton(entity), options);
+	}
+
 	/**
 	 * Add an array of updates to the batch.
 	 *
@@ -114,6 +149,22 @@ public interface CassandraBatchOperations {
 	 */
 	CassandraBatchOperations update(Iterable<?> entities, WriteOptions options);
 
+	/**
+	 * Add delete to the batch.
+	 *
+	 * @param entity the entity to delete; must not be {@literal null}.
+	 * @param options the WriteOptions to apply; must not be {@literal null}.
+	 * @return {@code this} {@link CassandraBatchOperations}.
+	 * @throws IllegalStateException if the batch was already executed.
+	 * @since 3.2.2
+	 */
+	default CassandraBatchOperations delete(Object entity, WriteOptions options) {
+
+		Assert.notNull(entity, "Entity must not be null");
+
+		return delete(Collections.singleton(entity), options);
+	}
+
 	/**
 	 * Add an array of deletes to the batch.
 	 *

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

@@ -20,6 +20,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.springframework.data.cassandra.core.convert.CassandraConverter;
 import org.springframework.data.cassandra.core.convert.UpdateMapper;
+import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.cassandra.core.cql.WriteOptions;
 import org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
@@ -154,9 +155,9 @@ class CassandraBatchTemplate implements CassandraBatchOperations {
 	public CassandraBatchOperations insert(Iterable<?> entities, WriteOptions options) {
 
 		assertNotExecuted();
-
 		Assert.notNull(entities, "Entities must not be null");
 		Assert.notNull(options, "WriteOptions must not be null");
+		assertNotQueryOptions(entities);
 
 		CassandraMappingContext mappingContext = getMappingContext();
 
@@ -202,9 +203,9 @@ class CassandraBatchTemplate implements CassandraBatchOperations {
 	public CassandraBatchOperations update(Iterable<?> entities, WriteOptions options) {
 
 		assertNotExecuted();
-
 		Assert.notNull(entities, "Entities must not be null");
 		Assert.notNull(options, "WriteOptions must not be null");
+		assertNotQueryOptions(entities);
 
 		for (Object entity : entities) {
 
@@ -247,9 +248,9 @@ class CassandraBatchTemplate implements CassandraBatchOperations {
 	public CassandraBatchOperations delete(Iterable<?> entities, WriteOptions options) {
 
 		assertNotExecuted();
-
 		Assert.notNull(entities, "Entities must not be null");
 		Assert.notNull(options, "WriteOptions must not be null");
+		assertNotQueryOptions(entities);
 
 		for (Object entity : entities) {
 
@@ -266,6 +267,17 @@ class CassandraBatchTemplate implements CassandraBatchOperations {
 		return this;
 	}
 
+	private void assertNotQueryOptions(Iterable<?> entities) {
+
+		for (Object entity : entities) {
+			if (entity instanceof QueryOptions) {
+				throw new IllegalArgumentException(
+						String.format("%s must not be used as entity. Please make sure to call the appropriate method accepting %s",
+								ClassUtils.getDescriptiveType(entity), ClassUtils.getShortName(entity.getClass())));
+			}
+		}
+	}
+
 	private void assertNotExecuted() {
 		Assert.state(!this.executed.get(), "This Cassandra Batch was already executed");
 	}

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

@@ -17,8 +17,12 @@ package org.springframework.data.cassandra.core;
 
 import reactor.core.publisher.Mono;
 
+import java.util.Collections;
+
 import org.reactivestreams.Subscriber;
+
 import org.springframework.data.cassandra.core.cql.WriteOptions;
+import org.springframework.util.Assert;
 
 /**
  * Reactive Batch operations for insert/update/delete actions on a table. {@link ReactiveCassandraBatchOperations} use
@@ -58,6 +62,22 @@ public interface ReactiveCassandraBatchOperations {
 	 */
 	ReactiveCassandraBatchOperations withTimestamp(long timestamp);
 
+	/**
+	 * Add an insert to the batch.
+	 *
+	 * @param entity the entity to insert; must not be {@literal null}.
+	 * @param options the WriteOptions to apply; must not be {@literal null}.
+	 * @return {@code this} {@link ReactiveCassandraBatchOperations}.
+	 * @throws IllegalStateException if the batch was already executed.
+	 * @since 3.2.2
+	 */
+	default ReactiveCassandraBatchOperations insert(Object entity, WriteOptions options) {
+
+		Assert.notNull(entity, "Entity must not be null");
+
+		return insert(Collections.singleton(entity), options);
+	}
+
 	/**
 	 * Add an array of inserts to the batch.
 	 *
@@ -107,6 +127,22 @@ public interface ReactiveCassandraBatchOperations {
 	 */
 	ReactiveCassandraBatchOperations insert(Mono<? extends Iterable<?>> entities, WriteOptions options);
 
+	/**
+	 * Add an update to the batch.
+	 *
+	 * @param entity the entity to update; must not be {@literal null}.
+	 * @param options the WriteOptions to apply; must not be {@literal null}.
+	 * @return {@code this} {@link ReactiveCassandraBatchOperations}.
+	 * @throws IllegalStateException if the batch was already executed.
+	 * @since 3.2.2
+	 */
+	default ReactiveCassandraBatchOperations update(Object entity, WriteOptions options) {
+
+		Assert.notNull(entity, "Entity must not be null");
+
+		return insert(Collections.singleton(entity), options);
+	}
+
 	/**
 	 * Add an array of updates to the batch.
 	 *
@@ -156,6 +192,22 @@ public interface ReactiveCassandraBatchOperations {
 	 */
 	ReactiveCassandraBatchOperations update(Mono<? extends Iterable<?>> entities, WriteOptions options);
 
+	/**
+	 * Add delete to the batch.
+	 *
+	 * @param entity the entity to delete; must not be {@literal null}.
+	 * @param options the WriteOptions to apply; must not be {@literal null}.
+	 * @return {@code this} {@link ReactiveCassandraBatchOperations}.
+	 * @throws IllegalStateException if the batch was already executed.
+	 * @since 3.2.2
+	 */
+	default ReactiveCassandraBatchOperations delete(Object entity, WriteOptions options) {
+
+		Assert.notNull(entity, "Entity must not be null");
+
+		return delete(Collections.singleton(entity), options);
+	}
+
 	/**
 	 * Add an array of deletes to the batch.
 	 *

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

@@ -15,6 +15,9 @@
  */
 package org.springframework.data.cassandra.core;
 
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -22,10 +25,9 @@ import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
 import org.springframework.data.cassandra.core.convert.CassandraConverter;
 import org.springframework.data.cassandra.core.convert.UpdateMapper;
+import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.cassandra.core.cql.WriteOptions;
 import org.springframework.data.cassandra.core.mapping.BasicCassandraPersistentEntity;
 import org.springframework.data.cassandra.core.mapping.CassandraMappingContext;
@@ -193,9 +195,9 @@ class ReactiveCassandraBatchTemplate implements ReactiveCassandraBatchOperations
 	public ReactiveCassandraBatchOperations insert(Iterable<?> entities, WriteOptions options) {
 
 		assertNotExecuted();
-
 		Assert.notNull(entities, "Entities must not be null");
 		Assert.notNull(options, "WriteOptions must not be null");
+		assertNotQueryOptions(entities);
 
 		this.batchMonos.add(Mono.just(doInsert(entities, options)));
 
@@ -209,7 +211,6 @@ class ReactiveCassandraBatchTemplate implements ReactiveCassandraBatchOperations
 	public ReactiveCassandraBatchOperations insert(Mono<? extends Iterable<?>> entities, WriteOptions options) {
 
 		assertNotExecuted();
-
 		Assert.notNull(entities, "Entities must not be null");
 		Assert.notNull(options, "WriteOptions must not be null");
 
@@ -273,9 +274,9 @@ class ReactiveCassandraBatchTemplate implements ReactiveCassandraBatchOperations
 	public ReactiveCassandraBatchOperations update(Iterable<?> entities, WriteOptions options) {
 
 		assertNotExecuted();
-
 		Assert.notNull(entities, "Entities must not be null");
 		Assert.notNull(options, "WriteOptions must not be null");
+		assertNotQueryOptions(entities);
 
 		this.batchMonos.add(Mono.just(doUpdate(entities, options)));
 
@@ -289,7 +290,6 @@ class ReactiveCassandraBatchTemplate implements ReactiveCassandraBatchOperations
 	public ReactiveCassandraBatchOperations update(Mono<? extends Iterable<?>> entities, WriteOptions options) {
 
 		assertNotExecuted();
-
 		Assert.notNull(entities, "Entities must not be null");
 		Assert.notNull(options, "WriteOptions must not be null");
 
@@ -351,9 +351,9 @@ class ReactiveCassandraBatchTemplate implements ReactiveCassandraBatchOperations
 	public ReactiveCassandraBatchOperations delete(Iterable<?> entities, WriteOptions options) {
 
 		assertNotExecuted();
-
 		Assert.notNull(entities, "Entities must not be null");
 		Assert.notNull(options, "WriteOptions must not be null");
+		assertNotQueryOptions(entities);
 
 		this.batchMonos.add(Mono.just(doDelete(entities, options)));
 
@@ -367,7 +367,6 @@ class ReactiveCassandraBatchTemplate implements ReactiveCassandraBatchOperations
 	public ReactiveCassandraBatchOperations delete(Mono<? extends Iterable<?>> entities, WriteOptions options) {
 
 		assertNotExecuted();
-
 		Assert.notNull(entities, "Entities must not be null");
 		Assert.notNull(options, "WriteOptions must not be null");
 
@@ -376,6 +375,17 @@ class ReactiveCassandraBatchTemplate implements ReactiveCassandraBatchOperations
 		return this;
 	}
 
+	private void assertNotQueryOptions(Iterable<?> entities) {
+
+		for (Object entity : entities) {
+			if (entity instanceof QueryOptions) {
+				throw new IllegalArgumentException(
+						String.format("%s must not be used as entity. Please make sure to call the appropriate method accepting %s",
+								ClassUtils.getDescriptiveType(entity), ClassUtils.getShortName(entity.getClass())));
+			}
+		}
+	}
+
 	private Collection<SimpleStatement> doDelete(Iterable<?> entities, WriteOptions options) {
 
 		List<SimpleStatement> deleteQueries = new ArrayList<>();

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

@@ -18,7 +18,6 @@ package org.springframework.data.cassandra.core;
 import static org.assertj.core.api.Assertions.*;
 
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.BeforeEach;
@@ -73,6 +72,12 @@ class CassandraBatchTemplateIntegrationTests extends AbstractKeyspaceCreatingInt
 		assertThat(loaded.getId().getUsername()).isEqualTo(walter.getId().getUsername());
 	}
 
+	@Test // #1135
+	void insertAsVarargsShouldRejectQueryOptions() {
+		assertThatIllegalArgumentException()
+				.isThrownBy(() -> template.batchOps().insert(mike, walter, InsertOptions.empty()));
+	}
+
 	@Test // DATACASS-288
 	void shouldInsertEntitiesWithLwt() {
 
@@ -85,7 +90,8 @@ class CassandraBatchTemplateIntegrationTests extends AbstractKeyspaceCreatingInt
 		walter.setAge(100);
 
 		CassandraBatchOperations batchOperations = new CassandraBatchTemplate(template);
-		WriteResult writeResult = batchOperations.insert(Collections.singleton(walter), lwtOptions).insert(mike).execute();
+
+		WriteResult writeResult = batchOperations.insert(walter, lwtOptions).insert(mike).execute();
 
 		Group loadedWalter = template.selectOneById(walter.getId(), Group.class);
 		Group loadedMike = template.selectOneById(mike.getId(), Group.class);
@@ -131,6 +137,12 @@ class CassandraBatchTemplateIntegrationTests extends AbstractKeyspaceCreatingInt
 		}
 	}
 
+	@Test // #1135
+	void updateAsVarargsShouldRejectQueryOptions() {
+		assertThatIllegalArgumentException()
+				.isThrownBy(() -> template.batchOps().update(mike, walter, InsertOptions.empty()));
+	}
+
 	@Test // DATACASS-288
 	void shouldUpdateEntities() {
 
@@ -169,14 +181,19 @@ class CassandraBatchTemplateIntegrationTests extends AbstractKeyspaceCreatingInt
 		WriteOptions options = WriteOptions.builder().ttl(ttl).build();
 
 		CassandraBatchOperations batchOperations = new CassandraBatchTemplate(template);
-		batchOperations.update(Arrays.asList(walter, mike), options).execute();
+		batchOperations.update(walter, options).execute();
 
-		ResultSet resultSet = template.getCqlOperations().queryForResultSet("SELECT TTL(email) FROM group;");
+		ResultSet resultSet = template.getCqlOperations().queryForResultSet("SELECT TTL(email), email FROM group");
 
 		assertThat(resultSet.getAvailableWithoutFetching()).isEqualTo(2);
 
 		for (Row row : resultSet) {
-			assertThat(row.getInt(0)).isBetween(1, ttl);
+
+			if (walter.getEmail().equals(row.getString("email"))) {
+				assertThat(row.getInt(0)).isBetween(1, ttl);
+			} else {
+				assertThat(row.getInt(0)).isZero();
+			}
 		}
 	}
 
@@ -200,6 +217,12 @@ class CassandraBatchTemplateIntegrationTests extends AbstractKeyspaceCreatingInt
 		assertThat(loaded.getEmail()).isEqualTo(walter.getEmail());
 	}
 
+	@Test // #1135
+	void deleteAsVarargsShouldRejectQueryOptions() {
+		assertThatIllegalArgumentException()
+				.isThrownBy(() -> template.batchOps().delete(mike, walter, InsertOptions.empty()));
+	}
+
 	@Test // DATACASS-288
 	void shouldDeleteEntities() {
 

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

@@ -28,6 +28,7 @@ import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+
 import org.springframework.data.cassandra.ReactiveResultSet;
 import org.springframework.data.cassandra.core.convert.MappingCassandraConverter;
 import org.springframework.data.cassandra.core.cql.ReactiveCqlTemplate;
@@ -39,6 +40,8 @@ import org.springframework.data.cassandra.domain.GroupKey;
 import org.springframework.data.cassandra.repository.support.SchemaTestUtils;
 import org.springframework.data.cassandra.test.util.AbstractKeyspaceCreatingIntegrationTests;
 
+import com.datastax.oss.driver.api.core.cql.Row;
+
 /**
  * Integration tests for {@link ReactiveCassandraBatchTemplate}.
  *
@@ -73,6 +76,12 @@ class ReactiveCassandraBatchTemplateIntegrationTests extends AbstractKeyspaceCre
 				.verifyComplete();
 	}
 
+	@Test // #1135
+	void insertAsVarargsShouldRejectQueryOptions() {
+		assertThatIllegalArgumentException()
+				.isThrownBy(() -> template.batchOps().insert(mike, walter, InsertOptions.empty()));
+	}
+
 	@Test // DATACASS-574
 	void shouldInsertEntities() {
 
@@ -104,19 +113,29 @@ class ReactiveCassandraBatchTemplateIntegrationTests extends AbstractKeyspaceCre
 	void shouldInsertCollectionOfEntitiesWithTtl() {
 
 		walter.setEmail("walter@white.com");
-		mike.setEmail("mike@sauls.com");
 
 		int ttl = 30;
 		WriteOptions options = WriteOptions.builder().ttl(30).build();
 
 		ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template);
-		Mono<ReactiveResultSet> resultSet = batchOperations.insert(Arrays.asList(walter, mike), options).execute()
-				.then(template.getReactiveCqlOperations().queryForResultSet("SELECT TTL(email) FROM group;"));
+		Mono<ReactiveResultSet> resultSet = batchOperations.insert(walter, options).execute()
+				.then(template.getReactiveCqlOperations().queryForResultSet("SELECT TTL(email), email FROM group;"));
 
 		resultSet.flatMapMany(ReactiveResultSet::availableRows) //
+				.collectList()
 				.as(StepVerifier::create) //
-				.assertNext(row -> assertThat(row.getInt(0)).isBetween(1, ttl))
-				.assertNext(row -> assertThat(row.getInt(0)).isBetween(1, ttl)).verifyComplete();
+				.assertNext(rows -> {
+
+					for (Row row : rows) {
+
+						if (walter.getEmail().equals(row.getString(1))) {
+							assertThat(row.getInt(0)).isBetween(1, ttl);
+						} else {
+							assertThat(row.getInt(0)).isZero();
+						}
+
+					}
+				}).verifyComplete();
 	}
 
 	@Test // DATACASS-574
@@ -138,6 +157,12 @@ class ReactiveCassandraBatchTemplateIntegrationTests extends AbstractKeyspaceCre
 				.assertNext(row -> assertThat(row.getInt(0)).isBetween(1, ttl)).verifyComplete();
 	}
 
+	@Test // #1135
+	void updateAsVarargsShouldRejectQueryOptions() {
+		assertThatIllegalArgumentException()
+				.isThrownBy(() -> template.batchOps().update(mike, walter, InsertOptions.empty()));
+	}
+
 	@Test // DATACASS-574
 	void shouldUpdateEntities() {
 
@@ -187,20 +212,29 @@ class ReactiveCassandraBatchTemplateIntegrationTests extends AbstractKeyspaceCre
 	void shouldUpdateCollectionOfEntitiesWithTtl() {
 
 		walter.setEmail("walter@white.com");
-		mike.setEmail("mike@sauls.com");
 
 		int ttl = 30;
 		WriteOptions options = WriteOptions.builder().ttl(ttl).build();
 
 		ReactiveCassandraBatchOperations batchOperations = new ReactiveCassandraBatchTemplate(template);
-		Mono<ReactiveResultSet> resultSet = batchOperations.update(Arrays.asList(walter, mike), options).execute()
-				.then(template.getReactiveCqlOperations().queryForResultSet("SELECT TTL(email) FROM group;"));
+		Mono<ReactiveResultSet> resultSet = batchOperations.update(walter, options).execute()
+				.then(template.getReactiveCqlOperations().queryForResultSet("SELECT TTL(email), email FROM group;"));
 
 		resultSet.flatMapMany(ReactiveResultSet::availableRows) //
+				.collectList()
 				.as(StepVerifier::create) //
+				.assertNext(rows -> {
 
-				.assertNext(row -> assertThat(row.getInt(0)).isBetween(1, ttl))
-				.assertNext(row -> assertThat(row.getInt(0)).isBetween(1, ttl)).verifyComplete();
+					for (Row row : rows) {
+
+						if (walter.getEmail().equals(row.getString(1))) {
+							assertThat(row.getInt(0)).isBetween(1, ttl);
+						} else {
+							assertThat(row.getInt(0)).isZero();
+						}
+
+					}
+				}).verifyComplete();
 	}
 
 	@Test // DATACASS-574
@@ -260,6 +294,12 @@ class ReactiveCassandraBatchTemplateIntegrationTests extends AbstractKeyspaceCre
 				.assertNext(loaded -> assertThat(loaded.getEmail()).isEqualTo(walter.getEmail())).verifyComplete();
 	}
 
+	@Test // #1135
+	void deleteAsVarargsShouldRejectQueryOptions() {
+		assertThatIllegalArgumentException()
+				.isThrownBy(() -> template.batchOps().delete(mike, walter, InsertOptions.empty()));
+	}
+
 	@Test // DATACASS-574
 	void shouldDeleteEntities() {