GH-2355 - Improve handling of non-distinct collections and already visited objects.

This commit started as a bunch of tests that should show which
self-referential relationships with optimistic locking are supported.

The bug in #2355 could be reproduced with the `saveAll` call: The
underlying issue bieng that one or more items had been visited at least
twice: The first time as related object, than as root. The next root of
course still be on the old version.

That has been fixed by keeping the state machine tracking visited items
for the duration of the `saveAll` calls around. Thus an related object
will not be updated twice.

For all of this to work in the most efficient way possible, a `java.util.Set` should
be used for related objects, along with a valid `equals/hashCode` pair. This
is however neglected.

By adding an additional check in the state machine falling back to ID extraction, we
can improve the user experience a lot and remove that need in many cases.

This commit fixes #2355.

Author: michael-simons

src/main/java/org/springframework/data/neo4j/core/Neo4jTemplate.java

@@ -316,7 +316,7 @@ private Object convertIdValues(@Nullable Neo4jPersistentProperty idProperty, Obj
 	@Override
 	public <T> T save(T instance) {
 
-		return saveImpl(instance, Collections.emptyList());
+		return saveImpl(instance, Collections.emptyList(), null);
 	}
 
 	@Override
@@ -336,7 +336,7 @@ public <T, R> R saveAs(T instance, Class<R> resultType) {
 		Collection<PropertyPath> pps = PropertyFilterSupport.addPropertiesFrom(instance.getClass(), resultType,
 				projectionFactory, neo4jMappingContext);
 
-		T savedInstance = saveImpl(instance, pps);
+		T savedInstance = saveImpl(instance, pps, null);
 		if (projectionInformation.isClosed()) {
 			return projectionFactory.createProjection(resultType, savedInstance);
 		}
@@ -348,7 +348,11 @@ public <T, R> R saveAs(T instance, Class<R> resultType) {
 				this.findById(propertyAccessor.getProperty(idProperty), savedInstance.getClass()).get());
 	}
 
-	private <T> T saveImpl(T instance, Collection<PropertyPath> includedProperties) {
+	private <T> T saveImpl(T instance, @Nullable Collection<PropertyPath> includedProperties, @Nullable NestedRelationshipProcessingStateMachine stateMachine) {
+
+		if (stateMachine != null && stateMachine.hasProcessedValue(instance)) {
+			return instance;
+		}
 
 		Neo4jPersistentEntity<?> entityMetaData = neo4jMappingContext.getRequiredPersistentEntity(instance.getClass());
 		boolean isEntityNew = entityMetaData.isNew(instance);
@@ -393,9 +397,17 @@ private <T> T saveImpl(T instance, Collection<PropertyPath> includedProperties)
 			propertyAccessor.setProperty(entityMetaData.getRequiredIdProperty(), internalId);
 		}
 		TemplateSupport.updateVersionPropertyIfPossible(entityMetaData, propertyAccessor, newOrUpdatedNode.get());
-		processRelations(entityMetaData, instance, internalId, propertyAccessor, isEntityNew, includeProperty);
 
-		return propertyAccessor.getBean();
+		if (stateMachine == null) {
+			stateMachine = new NestedRelationshipProcessingStateMachine(neo4jMappingContext, instance, internalId);
+		}
+
+		stateMachine.markValueAsProcessed(instance, internalId);
+		processRelations(entityMetaData, propertyAccessor, isEntityNew, stateMachine, includeProperty);
+
+		T bean = propertyAccessor.getBean();
+		stateMachine.markValueAsProcessedAs(instance, bean);
+		return bean;
 	}
 
 	@SuppressWarnings("unchecked")
@@ -447,7 +459,8 @@ private <T> List<T> saveAllImpl(Iterable<T> instances, List<PropertyPath> includ
 				|| entityMetaData.getDynamicLabelsProperty().isPresent()) {
 			log.debug("Saving entities using single statements.");
 
-			return entities.stream().map(e -> saveImpl(e, includedProperties)).collect(Collectors.toList());
+			NestedRelationshipProcessingStateMachine stateMachine = new NestedRelationshipProcessingStateMachine(neo4jMappingContext);
+			return entities.stream().map(e -> saveImpl(e, includedProperties, stateMachine)).collect(Collectors.toList());
 		}
 
 		class Tuple3<T> {
@@ -486,7 +499,7 @@ class Tuple3<T> {
 			Neo4jPersistentProperty idProperty = entityMetaData.getRequiredIdProperty();
 			Object id = convertIdValues(idProperty, propertyAccessor.getProperty(idProperty));
 			Long internalId = idToInternalIdMapping.get(id);
-			return processRelations(entityMetaData, t.originalInstance, internalId, propertyAccessor, t.wasNew, TemplateSupport.computeIncludePropertyPredicate(includedProperties, entityMetaData));
+			return this.<T>processRelations(entityMetaData, propertyAccessor, t.wasNew, new NestedRelationshipProcessingStateMachine(neo4jMappingContext, t.originalInstance, internalId), TemplateSupport.computeIncludePropertyPredicate(includedProperties, entityMetaData));
 		}).collect(Collectors.toList());
 	}
 
@@ -634,24 +647,34 @@ private <T> ExecutableQuery<T> createExecutableQuery(Class<T> domainType, @Nulla
 	 * Starts of processing of the relationships.
 	 *
 	 * @param neo4jPersistentEntity  The description of the instance to save
-	 * @param originalInstance       The original parent instance. It is paramount to pass in the original instance (prior
-	 *                               to generating the id and prior to eventually create new instances via the property accessor,
-	 *                               so that we can reliable stop traversing relationships.
 	 * @param parentPropertyAccessor The property accessor of the parent, to modify the relationships
 	 * @param isParentObjectNew      A flag if the parent was new
+	 * @param stateMachine           Initial state of entity processing
 	 * @param includeProperty        A predicate telling to include a relationship property or not
+	 * @param <T>                    The type of the object being initially processed
+	 * @return The owner of the relations being processed
 	 */
-	private <T> T processRelations(Neo4jPersistentEntity<?> neo4jPersistentEntity, T originalInstance, Long internalId,
-								   PersistentPropertyAccessor<?> parentPropertyAccessor,
-								   boolean isParentObjectNew, PropertyFilter includeProperty) {
+	private <T> T processRelations(
+			Neo4jPersistentEntity<?> neo4jPersistentEntity,
+			PersistentPropertyAccessor<?> parentPropertyAccessor,
+			boolean isParentObjectNew,
+			NestedRelationshipProcessingStateMachine stateMachine,
+			PropertyFilter includeProperty
+	) {
 
 		PropertyFilter.RelaxedPropertyPath startingPropertyPath = PropertyFilter.RelaxedPropertyPath.withRootType(neo4jPersistentEntity.getUnderlyingClass());
 		return processNestedRelations(neo4jPersistentEntity, parentPropertyAccessor, isParentObjectNew,
-				new NestedRelationshipProcessingStateMachine(originalInstance, internalId), includeProperty, startingPropertyPath);
+				stateMachine, includeProperty, startingPropertyPath);
 	}
 
-	private <T> T processNestedRelations(Neo4jPersistentEntity<?> sourceEntity, PersistentPropertyAccessor<?> propertyAccessor,
-										 boolean isParentObjectNew, NestedRelationshipProcessingStateMachine stateMachine, PropertyFilter includeProperty, PropertyFilter.RelaxedPropertyPath previousPath) {
+	private <T> T processNestedRelations(
+			Neo4jPersistentEntity<?> sourceEntity,
+			PersistentPropertyAccessor<?> propertyAccessor,
+			boolean isParentObjectNew,
+			NestedRelationshipProcessingStateMachine stateMachine,
+			PropertyFilter includeProperty,
+			PropertyFilter.RelaxedPropertyPath previousPath
+	) {
 
 		Object fromId = propertyAccessor.getProperty(sourceEntity.getRequiredIdProperty());
 
@@ -908,12 +931,13 @@ <T, R> List<R> doSave(Iterable<R> instances, Class<T> domainType) {
 		Collection<PropertyPath> pps = PropertyFilterSupport.addPropertiesFrom(domainType, resultType,
 				projectionFactory, neo4jMappingContext);
 
+		NestedRelationshipProcessingStateMachine stateMachine = new NestedRelationshipProcessingStateMachine(neo4jMappingContext);
 		List<R> results = new ArrayList<>();
 		for (R instance : instances) {
 			EntityFromDtoInstantiatingConverter<T> converter = new EntityFromDtoInstantiatingConverter<>(domainType, neo4jMappingContext);
 			T domainObject = converter.convert(instance);
 
-			T savedEntity = saveImpl(domainObject, pps);
+			T savedEntity = saveImpl(domainObject, pps, stateMachine);
 
 			@SuppressWarnings("unchecked")
 			R convertedBack = (R) new DtoInstantiatingConverter(resultType, neo4jMappingContext).convertDirectly(savedEntity);

src/main/java/org/springframework/data/neo4j/core/ReactiveNeo4jTemplate.java

@@ -300,7 +300,7 @@ private Object convertIdValues(@Nullable Neo4jPersistentProperty idProperty, Obj
 	@Override
 	public <T> Mono<T> save(T instance) {
 
-		return saveImpl(instance, Collections.emptyList());
+		return saveImpl(instance, Collections.emptyList(), null);
 	}
 
 	@Override
@@ -320,7 +320,7 @@ public <T, R> Mono<R> saveAs(T instance, Class<R> resultType) {
 		Collection<PropertyPath> pps = PropertyFilterSupport.addPropertiesFrom(instance.getClass(), resultType,
 				projectionFactory, neo4jMappingContext);
 
-		Mono<T> savingPublisher = saveImpl(instance, pps);
+		Mono<T> savingPublisher = saveImpl(instance, pps, null);
 		if (projectionInformation.isClosed()) {
 			return savingPublisher.map(savedInstance -> projectionFactory.createProjection(resultType, savedInstance));
 		}
@@ -346,22 +346,35 @@ <T, R> Flux<R> doSave(Iterable<R> instances, Class<T> domainType) {
 		Collection<PropertyPath> pps = PropertyFilterSupport.addPropertiesFrom(domainType, resultType,
 				projectionFactory, neo4jMappingContext);
 
+		NestedRelationshipProcessingStateMachine stateMachine = new NestedRelationshipProcessingStateMachine(neo4jMappingContext);
 		return Flux.fromIterable(instances)
 			.flatMap(instance -> {
 				EntityFromDtoInstantiatingConverter<T> converter = new EntityFromDtoInstantiatingConverter<>(domainType, neo4jMappingContext);
 				T domainObject = converter.convert(instance);
 
 				@SuppressWarnings("unchecked")
-				Mono<R> result = saveImpl(domainObject, pps)
+				Mono<R> result = saveImpl(domainObject, pps, stateMachine)
 						.map(savedEntity -> (R) new DtoInstantiatingConverter(resultType, neo4jMappingContext).convertDirectly(savedEntity));
 				return result;
 			});
 	}
 
-	private <T> Mono<T> saveImpl(T instance, @Nullable Collection<PropertyPath> includedProperties) {
+	private <T> Mono<T> saveImpl(T instance, @Nullable Collection<PropertyPath> includedProperties, @Nullable NestedRelationshipProcessingStateMachine stateMachine) {
+
+		if (stateMachine != null && stateMachine.hasProcessedValue(instance)) {
+			return Mono.just(instance);
+		}
 
 		Neo4jPersistentEntity<?> entityMetaData = neo4jMappingContext.getRequiredPersistentEntity(instance.getClass());
 		boolean isNewEntity = entityMetaData.isNew(instance);
+
+		NestedRelationshipProcessingStateMachine finalStateMachine;
+		if (stateMachine == null) {
+			finalStateMachine = new NestedRelationshipProcessingStateMachine(neo4jMappingContext);
+		} else {
+			finalStateMachine = stateMachine;
+		}
+
 		return Mono.just(instance).flatMap(eventSupport::maybeCallBeforeBind)
 				.flatMap(entityToBeSaved -> determineDynamicLabels(entityToBeSaved, entityMetaData)).flatMap(t -> {
 					T entityToBeSaved = t.getT1();
@@ -400,8 +413,9 @@ private <T> Mono<T> saveImpl(T instance, @Nullable Collection<PropertyPath> incl
 							propertyAccessor.setProperty(entityMetaData.getRequiredIdProperty(), newOrUpdatedNode.id());
 						}
 						TemplateSupport.updateVersionPropertyIfPossible(entityMetaData, propertyAccessor, newOrUpdatedNode);
+						finalStateMachine.markValueAsProcessed(instance, newOrUpdatedNode.id());
 					}).map(Entity::id)
-							.flatMap(internalId -> processRelations(entityMetaData, instance, internalId, propertyAccessor, isNewEntity, includeProperty));
+							.flatMap(internalId -> processRelations(entityMetaData,  propertyAccessor, isNewEntity, finalStateMachine, includeProperty));
 				});
 	}
 
@@ -486,7 +500,8 @@ private <T> Flux<T> saveAllImpl(Iterable<T> instances, @Nullable List<PropertyPa
 			|| entityMetaData.getDynamicLabelsProperty().isPresent()) {
 			log.debug("Saving entities using single statements.");
 
-			return Flux.fromIterable(entities).flatMap(e -> this.saveImpl(e, includedProperties));
+			NestedRelationshipProcessingStateMachine stateMachine = new NestedRelationshipProcessingStateMachine(neo4jMappingContext);
+			return Flux.fromIterable(entities).concatMap(e -> this.saveImpl(e, includedProperties, stateMachine));
 		}
 
 		@SuppressWarnings("unchecked") // We can safely assume here that we have a homongous collection with only one single type being either T or extending it
@@ -517,10 +532,8 @@ private <T> Flux<T> saveAllImpl(Iterable<T> instances, @Nullable List<PropertyPa
 							Neo4jPersistentProperty idProperty = entityMetaData.getRequiredIdProperty();
 							Object id = convertIdValues(idProperty, propertyAccessor.getProperty(idProperty));
 							Long internalId = idToInternalIdMapping.get(id);
-							return processRelations(entityMetaData, t.getT1(), internalId,
-									propertyAccessor, t.getT2(),
-								TemplateSupport.computeIncludePropertyPredicate(includedProperties,
-										entityMetaData));
+							return processRelations(entityMetaData, propertyAccessor, t.getT2(), new NestedRelationshipProcessingStateMachine(neo4jMappingContext, t.getT1(), internalId),
+								TemplateSupport.computeIncludePropertyPredicate(includedProperties, entityMetaData));
 						}))
 				);
 	}
@@ -750,22 +763,24 @@ Publisher<Tuple2<Collection<Long>, Collection<Long>>>> iterateAndMapNextLevel(
 	 * Starts of processing of the relationships.
 	 *
 	 * @param neo4jPersistentEntity  The description of the instance to save
-	 * @param originalInstance       The original parent instance. It is paramount to pass in the original instance (prior
-	 *                               to generating the id and prior to eventually create new instances via the property accessor,
-	 *                               so that we can reliable stop traversing relationships.
 	 * @param parentPropertyAccessor The property accessor of the parent, to modify the relationships
 	 * @param isParentObjectNew      A flag if the parent was new
+	 * @param stateMachine           Initial state of entity processing
 	 * @param includeProperty        A predicate telling to include a relationship property or not
-	 * @param <T>                    The type of the entity to save
-	 * @return A mono representing the whole stream of save operations.
+	 * @param <T>                    The type of the object being initially processed
+	 * @return A mono representing the whole stream of save operations, eventually containing the owner of the relations being processed
 	 */
-	private <T> Mono<T> processRelations(Neo4jPersistentEntity<?> neo4jPersistentEntity, T originalInstance,
-										 Long internalId, PersistentPropertyAccessor<?> parentPropertyAccessor,
-										 boolean isParentObjectNew, PropertyFilter includeProperty) {
+	private <T> Mono<T> processRelations(
+			Neo4jPersistentEntity<?> neo4jPersistentEntity,
+			PersistentPropertyAccessor<?> parentPropertyAccessor,
+			boolean isParentObjectNew,
+			NestedRelationshipProcessingStateMachine stateMachine,
+			PropertyFilter includeProperty
+	) {
 
 		PropertyFilter.RelaxedPropertyPath startingPropertyPath = PropertyFilter.RelaxedPropertyPath.withRootType(neo4jPersistentEntity.getUnderlyingClass());
 		return processNestedRelations(neo4jPersistentEntity, parentPropertyAccessor, isParentObjectNew,
-				new NestedRelationshipProcessingStateMachine(originalInstance, internalId), includeProperty, startingPropertyPath);
+				stateMachine, includeProperty, startingPropertyPath);
 	}
 
 	private <T> Mono<T> processNestedRelations(Neo4jPersistentEntity<?> sourceEntity, PersistentPropertyAccessor<?> parentPropertyAccessor,