Add mechanism for save to do one of insert, replace or upsert.

src/main/java/org/springframework/data/couchbase/repository/support/SimpleCouchbaseRepository.java

@@ -16,15 +16,15 @@
 
 package org.springframework.data.couchbase.repository.support;
 
-import static org.springframework.data.couchbase.repository.support.Util.hasNonZeroVersionProperty;
-
 import java.util.Collection;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.stream.Collectors;
 
 import org.springframework.data.couchbase.core.CouchbaseOperations;
+import org.springframework.data.couchbase.core.mapping.CouchbasePersistentEntity;
+import org.springframework.data.couchbase.core.mapping.CouchbasePersistentProperty;
 import org.springframework.data.couchbase.core.query.Query;
 import org.springframework.data.couchbase.repository.CouchbaseRepository;
 import org.springframework.data.couchbase.repository.query.CouchbaseEntityInformation;
@@ -35,6 +35,7 @@
 import org.springframework.data.util.StreamUtils;
 import org.springframework.data.util.Streamable;
 import org.springframework.util.Assert;
+import org.springframework.util.ReflectionUtils;
 
 import com.couchbase.client.java.query.QueryScanConsistency;
 
@@ -71,12 +72,25 @@ public SimpleCouchbaseRepository(CouchbaseEntityInformation<T, String> entityInf
 	@SuppressWarnings("unchecked")
 	public <S extends T> S save(S entity) {
 		Assert.notNull(entity, "Entity must not be null!");
-		// if entity has non-null, non-zero version property, then replace()
 		S result;
-		if (hasNonZeroVersionProperty(entity, operations.getConverter())) {
-			result = (S) operations.replaceById(getJavaType()).inScope(getScope()).inCollection(getCollection()).one(entity);
-		} else {
+
+		final CouchbasePersistentEntity<?> mapperEntity = operations.getConverter().getMappingContext()
+				.getPersistentEntity(entity.getClass());
+		final CouchbasePersistentProperty versionProperty = mapperEntity.getVersionProperty();
+		final boolean versionPresent = versionProperty != null;
+		final Long version = versionProperty == null || versionProperty.getField() == null ? null
+				: (Long) ReflectionUtils.getField(versionProperty.getField(), entity);
+		final boolean existingDocument = version != null && version > 0;
+
+		if (!versionPresent) { // the entity doesn't have a version property
+			// No version field - no cas
 			result = (S) operations.upsertById(getJavaType()).inScope(getScope()).inCollection(getCollection()).one(entity);
+		} else if (existingDocument) { // there is a version property, and it is non-zero
+			// Updating existing document with cas
+			result = (S) operations.replaceById(getJavaType()).inScope(getScope()).inCollection(getCollection()).one(entity);
+		} else { // there is a version property, but it's zero or not set.
+			// Creating new document
+			result = (S) operations.insertById(getJavaType()).inScope(getScope()).inCollection(getCollection()).one(entity);
 		}
 		return result;
 	}

src/main/java/org/springframework/data/couchbase/repository/support/SimpleReactiveCouchbaseRepository.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2021 the original author or authors.
+ * Copyright 2017-2022 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.
@@ -16,8 +16,6 @@
 
 package org.springframework.data.couchbase.repository.support;
 
-import static org.springframework.data.couchbase.repository.support.Util.hasNonZeroVersionProperty;
-
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -28,12 +26,15 @@
 import org.reactivestreams.Publisher;
 import org.springframework.data.couchbase.core.CouchbaseOperations;
 import org.springframework.data.couchbase.core.ReactiveCouchbaseOperations;
+import org.springframework.data.couchbase.core.mapping.CouchbasePersistentEntity;
+import org.springframework.data.couchbase.core.mapping.CouchbasePersistentProperty;
 import org.springframework.data.couchbase.core.query.Query;
 import org.springframework.data.couchbase.repository.ReactiveCouchbaseRepository;
 import org.springframework.data.couchbase.repository.query.CouchbaseEntityInformation;
 import org.springframework.data.domain.Sort;
 import org.springframework.data.util.Streamable;
 import org.springframework.util.Assert;
+import org.springframework.util.ReflectionUtils;
 
 /**
  * Reactive repository base implementation for Couchbase.
@@ -71,13 +72,26 @@ public SimpleReactiveCouchbaseRepository(CouchbaseEntityInformation<T, String> e
 	@Override
 	public <S extends T> Mono<S> save(S entity) {
 		Assert.notNull(entity, "Entity must not be null!");
-		// if entity has non-null, non-zero version property, then replace()
 		Mono<S> result;
-		if (hasNonZeroVersionProperty(entity, operations.getConverter())) {
+		final CouchbasePersistentEntity<?> mapperEntity = operations.getConverter().getMappingContext()
+				.getPersistentEntity(entity.getClass());
+		final CouchbasePersistentProperty versionProperty = mapperEntity.getVersionProperty();
+		final boolean versionPresent = versionProperty != null;
+		final Long version = versionProperty == null || versionProperty.getField() == null ? null
+				: (Long) ReflectionUtils.getField(versionProperty.getField(), entity);
+		final boolean existingDocument = version != null && version > 0;
+
+		if (!versionPresent) { // the entity doesn't have a version property
+			// No version field - no cas
+			result = (Mono<S>) operations.upsertById(getJavaType()).inScope(getScope()).inCollection(getCollection())
+					.one(entity);
+		} else if (existingDocument) { // there is a version property, and it is non-zero
+			// Updating existing document with cas
 			result = (Mono<S>) operations.replaceById(getJavaType()).inScope(getScope()).inCollection(getCollection())
 					.one(entity);
-		} else {
-			result = (Mono<S>) operations.upsertById(getJavaType()).inScope(getScope()).inCollection(getCollection())
+		} else { // there is a version property, but it's zero or not set.
+			// Creating new document
+			result = (Mono<S>) operations.insertById(getJavaType()).inScope(getScope()).inCollection(getCollection())
 					.one(entity);
 		}
 		return result;

src/test/java/org/springframework/data/couchbase/domain/ReactiveAirlineRepository.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2012-2020 the original author or authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.data.couchbase.domain;
+
+import org.springframework.data.couchbase.repository.Query;
+import org.springframework.data.repository.query.Param;
+import reactor.core.publisher.Flux;
+
+import org.springframework.data.repository.reactive.ReactiveSortingRepository;
+import org.springframework.stereotype.Repository;
+
+import java.util.List;
+
+/**
+ * @author Michael Reiche
+ */
+@Repository
+public interface ReactiveAirlineRepository extends ReactiveSortingRepository<Airline, String> {
+
+	@Query("#{#n1ql.selectEntity} where #{#n1ql.filter} and (name = $1)")
+	List<User> getByName(@Param("airline_name")String airlineName);
+
+}

src/test/java/org/springframework/data/couchbase/repository/CouchbaseRepositoryKeyValueIntegrationTests.java

@@ -16,9 +16,12 @@
 
 package org.springframework.data.couchbase.repository;
 
+
+import static com.couchbase.client.java.query.QueryScanConsistency.REQUEST_PLUS;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.lang.reflect.InvocationTargetException;
@@ -28,11 +31,16 @@
 import java.util.Optional;
 import java.util.UUID;
 
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.dao.DuplicateKeyException;
 import org.springframework.data.couchbase.config.AbstractCouchbaseConfiguration;
 import org.springframework.data.couchbase.core.CouchbaseTemplate;
+import org.springframework.data.couchbase.domain.Airline;
+import org.springframework.data.couchbase.domain.AirlineRepository;
 import org.springframework.data.couchbase.domain.Course;
 import org.springframework.data.couchbase.domain.Library;
 import org.springframework.data.couchbase.domain.LibraryRepository;
@@ -67,9 +75,17 @@ public class CouchbaseRepositoryKeyValueIntegrationTests extends ClusterAwareInt
 	@Autowired LibraryRepository libraryRepository;
 	@Autowired SubscriptionTokenRepository subscriptionTokenRepository;
 	@Autowired UserSubmissionRepository userSubmissionRepository;
+	@Autowired AirlineRepository airlineRepository;
 	@Autowired PersonValueRepository personValueRepository;
 	@Autowired CouchbaseTemplate couchbaseTemplate;
 
+	@BeforeEach
+	public void beforeEach() {
+		super.beforeEach();
+		couchbaseTemplate.removeByQuery(SubscriptionToken.class).withConsistency(REQUEST_PLUS).all();
+		couchbaseTemplate.findByQuery(SubscriptionToken.class).withConsistency(REQUEST_PLUS).all();
+	}
+
 	@Test
 	void subscriptionToken() {
 		SubscriptionToken st = new SubscriptionToken("id", 0, "type", "Dave Smith", "app123", "dev123", 0);
@@ -82,6 +98,29 @@ void subscriptionToken() {
 		subscriptionTokenRepository.delete(st);
 	}
 
+	@Test
+	@IgnoreWhen(clusterTypes = ClusterType.MOCKED)
+	void saveReplaceUpsertInsert() {
+		// the User class has a version.
+		User user = new User(UUID.randomUUID().toString(), "f", "l");
+		// save the document - we don't care how on this call
+		userRepository.save(user);
+		// Now set the version to 0, it should attempt an insert and fail.
+		long saveVersion = user.getVersion();
+		user.setVersion(0);
+		assertThrows(DuplicateKeyException.class, () -> userRepository.save(user));
+		user.setVersion(saveVersion + 1);
+		assertThrows(DataIntegrityViolationException.class, () -> userRepository.save(user));
+		userRepository.delete(user);
+
+		// Airline does not have a version
+		Airline airline = new Airline(UUID.randomUUID().toString(), "MyAirline");
+		// save the document - we don't care how on this call
+		airlineRepository.save(airline);
+		airlineRepository.save(airline); // If it was an insert it would fail. Can't tell if it is an upsert or replace.
+		airlineRepository.delete(airline);
+	}
+
 	@Test
 	@IgnoreWhen(clusterTypes = ClusterType.MOCKED)
 	void saveAndFindById() {

src/test/java/org/springframework/data/couchbase/repository/CouchbaseRepositoryQueryIntegrationTests.java

@@ -48,6 +48,7 @@
 
 import javax.validation.ConstraintViolationException;
 
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
@@ -104,7 +105,7 @@
 import com.couchbase.client.java.env.ClusterEnvironment;
 import com.couchbase.client.java.json.JsonArray;
 import com.couchbase.client.java.kv.GetResult;
-import com.couchbase.client.java.kv.UpsertOptions;
+import com.couchbase.client.java.kv.InsertOptions;
 import com.couchbase.client.java.query.QueryOptions;
 import com.couchbase.client.java.query.QueryScanConsistency;
 
@@ -132,6 +133,13 @@ public class CouchbaseRepositoryQueryIntegrationTests extends ClusterAwareIntegr
 	String scopeName = "_default";
 	String collectionName = "_default";
 
+	@BeforeEach
+	public void beforeEach() {
+		super.beforeEach();
+		couchbaseTemplate.removeByQuery(User.class).withConsistency(REQUEST_PLUS).all();
+		couchbaseTemplate.findByQuery(User.class).withConsistency(REQUEST_PLUS).all();
+	}
+
 	@Test
 	void shouldSaveAndFindAll() {
 		Airport vie = null;
@@ -556,17 +564,18 @@ public void testTransient() {
 	public void testCas() {
 		User user = new User("1", "Dave", "Wilson");
 		userRepository.save(user);
+		long saveVersion = user.getVersion();
 		user.setVersion(user.getVersion() - 1);
 		assertThrows(DataIntegrityViolationException.class, () -> userRepository.save(user));
-		user.setVersion(0);
+		user.setVersion(saveVersion);
 		userRepository.save(user);
 		userRepository.delete(user);
 	}
 
 	@Test
 	public void testExpiration() {
 		Airport airport = new Airport("1", "iata21", "icao21");
-		airportRepository.withOptions(UpsertOptions.upsertOptions().expiry(Duration.ofSeconds(10))).save(airport);
+		airportRepository.withOptions(InsertOptions.insertOptions().expiry(Duration.ofSeconds(10))).save(airport);
 		Airport foundAirport = airportRepository.findByIata(airport.getIata());
 		assertNotEquals(0, foundAirport.getExpiration());
 		airportRepository.delete(airport);

src/test/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseRepositoryKeyValueIntegrationTests.java

@@ -18,6 +18,7 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.util.Optional;
@@ -27,9 +28,13 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.dao.DuplicateKeyException;
 import org.springframework.data.auditing.DateTimeProvider;
 import org.springframework.data.couchbase.config.AbstractCouchbaseConfiguration;
+import org.springframework.data.couchbase.domain.Airline;
 import org.springframework.data.couchbase.domain.Airport;
+import org.springframework.data.couchbase.domain.ReactiveAirlineRepository;
 import org.springframework.data.couchbase.domain.ReactiveAirportRepository;
 import org.springframework.data.couchbase.domain.ReactiveNaiveAuditorAware;
 import org.springframework.data.couchbase.domain.ReactiveUserRepository;
@@ -53,6 +58,31 @@ public class ReactiveCouchbaseRepositoryKeyValueIntegrationTests extends Cluster
 
 	@Autowired ReactiveAirportRepository airportRepository;
 
+	@Autowired ReactiveAirlineRepository airlineRepository;
+
+	@Test
+	@IgnoreWhen(clusterTypes = ClusterType.MOCKED)
+	void saveReplaceUpsertInsert() {
+		// the User class has a version.
+		User user = new User(UUID.randomUUID().toString(), "f", "l");
+		// save the document - we don't care how on this call
+		userRepository.save(user).block();
+		// Now set the version to 0, it should attempt an insert and fail.
+		long saveVersion = user.getVersion();
+		user.setVersion(0);
+		assertThrows(DuplicateKeyException.class, () -> userRepository.save(user).block());
+		user.setVersion(saveVersion + 1);
+		assertThrows(DataIntegrityViolationException.class, () -> userRepository.save(user).block());
+		userRepository.delete(user);
+
+		// Airline does not have a version
+		Airline airline = new Airline(UUID.randomUUID().toString(), "MyAirline");
+		// save the document - we don't care how on this call
+		airlineRepository.save(airline).block();
+		airlineRepository.save(airline).block(); // If it was an insert it would fail. Can't tell if an upsert or replace.
+		airlineRepository.delete(airline).block();
+	}
+
 	@Test
 	void saveAndFindById() {
 		User user = new User(UUID.randomUUID().toString(), "saveAndFindById_reactive", "l");

src/test/java/org/springframework/data/couchbase/repository/ReactiveCouchbaseRepositoryQueryIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2021 the original author or authors.
+ * Copyright 2017-2022 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.
@@ -118,9 +118,10 @@ void findBySimpleProperty() {
 	public void testCas() {
 		User user = new User("1", "Dave", "Wilson");
 		userRepository.save(user).block();
+		long saveVersion = user.getVersion();
 		user.setVersion(user.getVersion() - 1);
 		assertThrows(DataIntegrityViolationException.class, () -> userRepository.save(user).block());
-		user.setVersion(0);
+		user.setVersion(saveVersion);
 		userRepository.save(user).block();
 		userRepository.delete(user).block();
 	}