DATACASS-445 - Insert entire object on CassandraRepository.save(…).

We now insert the entire object including null values again in Cassandra using the repository save method. Objects that are saved through the repository are expected to be returned the same when retrieved again.

Cassandra entities don't have any indicator whether these entities are new to optimize null value handling and prevent tombstones in Cassandra. CassandraRepository exposes insert(…) to insert objects without inserting null values.

Previously, we attempted to execute insert/update conditionally on the entity structure which ended up issuing update statements containing all null values. Entities were not created and save(…) failed silently.

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/convert/MappingCassandraConverter.java

@@ -24,6 +24,7 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Map.Entry;
 
 import org.slf4j.Logger;
@@ -304,6 +305,7 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 	}
 
 	@Override
+	@SuppressWarnings("unchecked")
 	public void write(Object source, Object sink, CassandraPersistentEntity<?> entity) {
 
 		if (source == null) {
@@ -314,7 +316,9 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 			throw new MappingException("No mapping metadata found for " + source.getClass());
 		}
 
-		if (sink instanceof Insert) {
+		if (sink instanceof Map) {
+			writeMapFromWrapper(getConvertingAccessor(source, entity), (Map<String, Object>) sink, entity);
+		} else if (sink instanceof Insert) {
 			writeInsertFromObject(source, (Insert) sink, entity);
 		} else if (sink instanceof Update) {
 			writeUpdateFromObject(source, (Update) sink, entity);
@@ -333,6 +337,40 @@ public class MappingCassandraConverter extends AbstractCassandraConverter
 		writeInsertFromWrapper(getConvertingAccessor(object, entity), insert, entity);
 	}
 
+	private void writeMapFromWrapper(final ConvertingPropertyAccessor accessor, final Map<String, Object> insert,
+			CassandraPersistentEntity<?> entity) {
+
+		entity.doWithProperties(new PropertyHandler<CassandraPersistentProperty>() {
+
+			@Override
+			public void doWithPersistentProperty(CassandraPersistentProperty property) {
+
+				Object value = getWriteValue(property, accessor);
+
+				if (log.isDebugEnabled()) {
+					log.debug("doWithProperties Property.type {}, Property.value {}", property.getType().getName(), value);
+				}
+
+				if (property.isCompositePrimaryKey()) {
+					if (log.isDebugEnabled()) {
+						log.debug("Property is a compositeKey");
+					}
+
+					writeMapFromWrapper(getConvertingAccessor(value, property.getCompositePrimaryKeyEntity()), insert,
+							property.getCompositePrimaryKeyEntity());
+
+					return;
+				}
+
+				if (log.isDebugEnabled()) {
+					log.debug("Adding map.entry [{}] - [{}]", property.getColumnName().toCql(), value);
+				}
+
+				insert.put(property.getColumnName().toCql(), value);
+			}
+		});
+	}
+
 	protected void writeInsertFromWrapper(final ConvertingPropertyAccessor accessor, final Insert insert,
 			CassandraPersistentEntity<?> entity) {
 

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepository.java

@@ -16,15 +16,24 @@
 package org.springframework.data.cassandra.repository.support;
 
 import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
 
 import org.springframework.cassandra.core.util.CollectionUtils;
+import org.springframework.data.cassandra.convert.CassandraConverter;
 import org.springframework.data.cassandra.core.CassandraOperations;
 import org.springframework.data.cassandra.core.CassandraTemplate;
+import org.springframework.data.cassandra.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.repository.TypedIdCassandraRepository;
 import org.springframework.data.cassandra.repository.query.CassandraEntityInformation;
 import org.springframework.util.Assert;
 
+import com.datastax.driver.core.querybuilder.Batch;
+import com.datastax.driver.core.querybuilder.Insert;
+import com.datastax.driver.core.querybuilder.QueryBuilder;
 import com.datastax.driver.core.querybuilder.Select;
 
 /**
@@ -39,8 +48,6 @@ public class SimpleCassandraRepository<T, ID extends Serializable> implements Ty
 	protected CassandraOperations operations;
 	protected CassandraEntityInformation<T, ID> entityInformation;
 
-	private final boolean isPrimaryKeyEntity;
-
 	/**
 	 * Creates a new {@link SimpleCassandraRepository} for the given {@link CassandraEntityInformation} and
 	 * {@link CassandraTemplate}.
@@ -55,7 +62,6 @@ public class SimpleCassandraRepository<T, ID extends Serializable> implements Ty
 
 		this.entityInformation = metadata;
 		this.operations = operations;
-		this.isPrimaryKeyEntity = metadata.isPrimaryKeyEntity();
 	}
 
 	/* (non-Javadoc)
@@ -66,11 +72,11 @@ public class SimpleCassandraRepository<T, ID extends Serializable> implements Ty
 
 		Assert.notNull(entity, "Entity must not be null");
 
-		if (entityInformation.isNew(entity) || isPrimaryKeyEntity) {
-			return operations.insert(entity);
-		}
+		Insert insert = createFullInsert(entity);
 
-		return operations.update(entity);
+		operations.execute(insert);
+
+		return entity;
 	}
 
 	/* (non-Javadoc)
@@ -81,7 +87,36 @@ public class SimpleCassandraRepository<T, ID extends Serializable> implements Ty
 
 		Assert.notNull(entities, "The given Iterable of entities must not be null");
 
-		return operations.insert(CollectionUtils.toList(entities));
+		List<S> result = new ArrayList<S>();
+
+		Batch batch = QueryBuilder.batch();
+		for (S entity : entities) {
+
+			result.add(entity);
+			batch.add(createFullInsert(entity));
+		}
+
+		operations.execute(batch);
+
+		return result;
+	}
+
+	private <S extends T> Insert createFullInsert(S entity) {
+
+		CassandraConverter converter = operations.getConverter();
+		CassandraPersistentEntity<?> persistentEntity = converter.getMappingContext()
+				.getPersistentEntity(entity.getClass());
+		Map<String, Object> toInsert = new LinkedHashMap<String, Object>();
+
+		converter.write(entity, toInsert, persistentEntity);
+
+		Insert insert = QueryBuilder.insertInto(persistentEntity.getTableName().toCql());
+
+		for (Entry<String, Object> entry : toInsert.entrySet()) {
+			insert.value(entry.getKey(), entry.getValue());
+		}
+
+		return insert;
 	}
 
 	/* (non-Javadoc)

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/domain/Person.java

@@ -15,7 +15,9 @@
  */
 package org.springframework.data.cassandra.domain;
 
+import lombok.AllArgsConstructor;
 import lombok.Data;
+import lombok.NoArgsConstructor;
 
 import org.springframework.data.annotation.Id;
 import org.springframework.data.cassandra.mapping.Table;
@@ -25,6 +27,8 @@ import org.springframework.data.cassandra.mapping.Table;
  */
 @Table
 @Data
+@AllArgsConstructor
+@NoArgsConstructor
 public class Person {
 
 	@Id String id;

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepositoryIntegrationTests.java

@@ -0,0 +1,301 @@
+/*
+ * Copyright 2017 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.repository.support;
+
+import static org.assertj.core.api.Assertions.*;
+
+import java.util.Arrays;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanClassLoaderAware;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cassandra.test.integration.AbstractKeyspaceCreatingIntegrationTest;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.cassandra.core.CassandraOperations;
+import org.springframework.data.cassandra.domain.Person;
+import org.springframework.data.cassandra.repository.TypedIdCassandraRepository;
+import org.springframework.data.cassandra.test.integration.support.IntegrationTestConfig;
+import org.springframework.data.repository.query.DefaultEvaluationContextProvider;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * Integration tests for {@link SimpleCassandraRepository}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration
+public class SimpleCassandraRepositoryIntegrationTests extends AbstractKeyspaceCreatingIntegrationTest
+		implements BeanClassLoaderAware, BeanFactoryAware {
+
+	@Configuration
+	public static class Config extends IntegrationTestConfig {
+
+		@Override
+		public String[] getEntityBasePackages() {
+			return new String[] { Person.class.getPackage().getName() };
+		}
+	}
+
+	@Autowired private CassandraOperations operations;
+
+	CassandraRepositoryFactory factory;
+	ClassLoader classLoader;
+	BeanFactory beanFactory;
+	PersonRepostitory repository;
+
+	Person dave, oliver, carter, boyd;
+
+	@Override
+	public void setBeanClassLoader(ClassLoader classLoader) {
+		this.classLoader = classLoader == null ? org.springframework.util.ClassUtils.getDefaultClassLoader() : classLoader;
+	}
+
+	@Override
+	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
+		this.beanFactory = beanFactory;
+	}
+
+	@Before
+	public void setUp() {
+
+		factory = new CassandraRepositoryFactory(operations);
+		factory.setRepositoryBaseClass(SimpleCassandraRepository.class);
+		factory.setBeanClassLoader(classLoader);
+		factory.setBeanFactory(beanFactory);
+		factory.setEvaluationContextProvider(DefaultEvaluationContextProvider.INSTANCE);
+
+		repository = factory.getRepository(PersonRepostitory.class);
+
+		repository.deleteAll();
+
+		dave = new Person("42", "Dave", "Matthews");
+		oliver = new Person("4", "Oliver August", "Matthews");
+		carter = new Person("49", "Carter", "Beauford");
+		boyd = new Person("45", "Boyd", "Tinsley");
+
+		repository.save(Arrays.asList(oliver, dave, carter, boyd));
+	}
+
+	@Test // DATACASS-445
+	public void existsByIdShouldReturnTrueForExistingObject() {
+
+		Boolean exists = repository.exists(dave.getId());
+
+		assertThat(exists).isTrue();
+	}
+
+	@Test // DATACASS-445
+	public void existsByIdShouldReturnFalseForAbsentObject() {
+
+		boolean exists = repository.exists("unknown");
+
+		assertThat(exists).isFalse();
+	}
+
+	@Test // DATACASS-445
+	public void existsByMonoOfIdShouldReturnTrueForExistingObject() {
+
+		boolean exists = repository.exists(dave.getId());
+
+		assertThat(exists).isTrue();
+	}
+
+	@Test // DATACASS-445
+	public void findOneShouldReturnObject() {
+
+		Person person = repository.findOne(dave.getId());
+
+		assertThat(person).isEqualTo(dave);
+	}
+
+	@Test // DATACASS-445
+	public void findOneShouldCompleteWithoutValueForAbsentObject() {
+
+		Person person = repository.findOne("unknown");
+
+		assertThat(person).isNull();
+	}
+
+	@Test // DATACASS-445
+	public void findAllShouldReturnAllResults() {
+
+		Iterable<Person> persons = repository.findAll();
+
+		assertThat(persons).hasSize(4);
+	}
+
+	@Test // DATACASS-445
+	public void findAllByIterableOfIdShouldReturnResults() {
+
+		Iterable<Person> persons = repository.findAll(Arrays.asList(dave.getId(), boyd.getId()));
+
+		assertThat(persons).hasSize(2);
+	}
+
+	@Test // DATACASS-445
+	public void countShouldReturnNumberOfRecords() {
+
+		long count = repository.count();
+
+		assertThat(count).isEqualTo(4);
+	}
+
+	@Test // DATACASS-415
+	public void insertEntityShouldInsertEntity() {
+
+		repository.deleteAll();
+
+		Person person = new Person("36", "Homer", "Simpson");
+
+		repository.insert(person);
+
+		assertThat(repository.count()).isEqualTo(1);
+	}
+
+	@Test // DATACASS-415
+	public void insertIterableOfEntitiesShouldInsertEntity() {
+
+		repository.deleteAll();
+
+		repository.save(Arrays.asList(dave, oliver, boyd));
+
+		assertThat(repository.count()).isEqualTo(3);
+	}
+
+	@Test // DATACASS-445
+	public void saveEntityShouldUpdateExistingEntity() {
+
+		dave.setFirstname("Hello, Dave");
+		dave.setLastname("Bowman");
+
+		Person saved = repository.save(dave);
+
+		assertThat(saved).isEqualTo(saved);
+
+		Person loaded = repository.findOne(dave.getId());
+
+		assertThat(loaded.getFirstname()).isEqualTo(dave.getFirstname());
+		assertThat(loaded.getLastname()).isEqualTo(dave.getLastname());
+	}
+
+	@Test // DATACASS-445
+	public void saveEntityShouldInsertPartialEntity() {
+
+		Person justId = new Person("foo", null, null);
+		Person saved = repository.save(justId);
+
+		assertThat(saved).isEqualTo(saved);
+
+		Person loaded = repository.findOne(justId.getId());
+
+		assertThat(loaded.getFirstname()).isNull();
+		assertThat(loaded.getLastname()).isNull();
+	}
+
+	@Test // DATACASS-445
+	public void saveEntityShouldInsertNewEntity() {
+
+		Person person = new Person("36", "Homer", "Simpson");
+
+		Person saved = repository.save(person);
+
+		assertThat(saved).isEqualTo(person);
+
+		Person loaded = repository.findOne(person.getId());
+
+		assertThat(loaded).isEqualTo(person);
+	}
+
+	@Test // DATACASS-445
+	public void saveIterableOfNewEntitiesShouldInsertEntity() {
+
+		repository.deleteAll();
+
+		Iterable<Person> saved = repository.save(Arrays.asList(dave, oliver, boyd));
+
+		assertThat(saved).hasSize(3);
+
+		assertThat(repository.count()).isEqualTo(3);
+	}
+
+	@Test // DATACASS-445
+	public void saveIterableOfMixedEntitiesShouldInsertEntity() {
+
+		Person person = new Person("36", "Homer", "Simpson");
+
+		dave.setFirstname("Hello, Dave");
+		dave.setLastname("Bowman");
+
+		Iterable<Person> saved = repository.save(Arrays.asList(person, dave));
+
+		assertThat(saved).hasSize(2);
+
+		Person persistentDave = repository.findOne(dave.getId());
+		assertThat(persistentDave).isEqualTo(dave);
+
+		Person persistentHomer = repository.findOne(person.getId());
+		assertThat(persistentHomer).isEqualTo(person);
+	}
+
+	@Test // DATACASS-445
+	public void deleteAllShouldRemoveEntities() {
+
+		repository.deleteAll();
+
+		Iterable<Person> result = repository.findAll();
+
+		assertThat(result).isEmpty();
+	}
+
+	@Test // DATACASS-445
+	public void deleteByIdShouldRemoveEntity() {
+
+		repository.delete(dave.getId());
+
+		Person loaded = repository.findOne(dave.getId());
+
+		assertThat(loaded).isNull();
+	}
+
+	@Test // DATACASS-445
+	public void deleteShouldRemoveEntity() {
+
+		repository.delete(dave);
+
+		Person loaded = repository.findOne(dave.getId());
+
+		assertThat(loaded).isNull();
+	}
+
+	@Test // DATACASS-445
+	public void deleteIterableOfEntitiesShouldRemoveEntities() {
+
+		repository.delete(Arrays.asList(dave, boyd));
+
+		Person loaded = repository.findOne(boyd.getId());
+
+		assertThat(loaded).isNull();
+	}
+
+	interface PersonRepostitory extends TypedIdCassandraRepository<Person, String> {}
+}

spring-data-cassandra/src/test/java/org/springframework/data/cassandra/repository/support/SimpleCassandraRepositoryUnitTests.java

@@ -25,8 +25,11 @@ import java.io.Serializable;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
 import org.mockito.Mock;
 import org.mockito.runners.MockitoJUnitRunner;
+import org.springframework.cassandra.core.cql.CqlIdentifier;
 import org.springframework.data.annotation.Id;
 import org.springframework.data.cassandra.convert.MappingCassandraConverter;
 import org.springframework.data.cassandra.core.CassandraOperations;
@@ -35,6 +38,9 @@ import org.springframework.data.cassandra.mapping.CassandraPersistentEntity;
 import org.springframework.data.cassandra.mapping.UserTypeResolver;
 import org.springframework.data.cassandra.test.integration.repository.querymethods.declared.Person;
 
+import com.datastax.driver.core.UserType;
+import com.datastax.driver.core.querybuilder.Insert;
+
 /**
  * Unit tests for {@link SimpleCassandraRepository}.
  *
@@ -51,13 +57,20 @@ public class SimpleCassandraRepositoryUnitTests {
 
 	@Mock CassandraOperations cassandraOperations;
 	@Mock UserTypeResolver userTypeResolver;
+	@Mock UserType userType;
+
+	@Captor ArgumentCaptor<Insert> insertCaptor;
 
 	@Before
 	public void before() {
+
 		mappingContext.setUserTypeResolver(userTypeResolver);
+
+		when(cassandraOperations.getConverter()).thenReturn(converter);
+		when(userTypeResolver.resolveType(CqlIdentifier.cqlId("address"))).thenReturn(userType);
 	}
 
-	@Test // DATACASS-428
+	@Test // DATACASS-428, DATACASS-445
 	public void saveShouldInsertNewPrimaryKeyOnlyEntity() {
 
 		CassandraPersistentEntity<?> entity = converter.getMappingContext().getPersistentEntity(SimplePerson.class);
@@ -67,15 +80,13 @@ public class SimpleCassandraRepositoryUnitTests {
 
 		SimplePerson person = new SimplePerson();
 
-		when(cassandraOperations.insert(person)).thenReturn(person);
+		repository.save(person);
 
-		Object result = repository.save(person);
-
-		assertThat(result).isEqualTo(person);
-		verify(cassandraOperations).insert(person);
+		verify(cassandraOperations).execute(insertCaptor.capture());
+		assertThat(insertCaptor.getValue().toString()).isEqualTo("INSERT INTO simpleperson (id) VALUES (null);");
 	}
 
-	@Test // DATACASS-428
+	@Test // DATACASS-428, DATACASS-445
 	public void saveShouldUpdateNewEntity() {
 
 		CassandraPersistentEntity<?> entity = converter.getMappingContext().getPersistentEntity(Person.class);
@@ -85,15 +96,12 @@ public class SimpleCassandraRepositoryUnitTests {
 
 		Person person = new Person();
 
-		when(cassandraOperations.update(person)).thenReturn(person);
+		repository.save(person);
 
-		Object result = repository.save(person);
-
-		assertThat(result).isEqualTo(person);
-		verify(cassandraOperations).update(person);
+		verify(cassandraOperations).execute(any(Insert.class));
 	}
 
-	@Test // DATACASS-428
+	@Test // DATACASS-428, DATACASS-445
 	public void saveShouldUpdateExistingEntity() {
 
 		CassandraPersistentEntity<?> entity = converter.getMappingContext().getPersistentEntity(Person.class);
@@ -105,15 +113,15 @@ public class SimpleCassandraRepositoryUnitTests {
 		person.setFirstname("foo");
 		person.setLastname("bar");
 
-		when(cassandraOperations.update(person)).thenReturn(person);
+		repository.save(person);
 
-		Object result = repository.save(person);
-
-		assertThat(result).isEqualTo(person);
-		verify(cassandraOperations).update(person);
+		verify(cassandraOperations).execute(insertCaptor.capture());
+		assertThat(insertCaptor.getValue().toString())
+				.contains("INSERT INTO person (lastname,firstname,alternativeaddresses,");
+		assertThat(insertCaptor.getValue().toString()).contains("VALUES ('bar','foo',null");
 	}
 
-	@Test // DATACASS-428
+	@Test // DATACASS-428, DATACASS-445
 	public void insertShouldInsertEntity() {
 
 		CassandraPersistentEntity<?> entity = converter.getMappingContext().getPersistentEntity(Person.class);
@@ -123,12 +131,9 @@ public class SimpleCassandraRepositoryUnitTests {
 
 		Person person = new Person();
 
-		when(cassandraOperations.insert(person)).thenReturn(person);
+		repository.insert(person);
 
-		Object result = repository.insert(person);
-
-		assertThat(result).isEqualTo(person);
-		verify(cassandraOperations).insert(person);
+		verify(cassandraOperations).insert(any(Insert.class));
 	}
 
 	@Data