Added Sub-Document mutations support (#1660)

Closes #1659

src/main/asciidoc/template.adoc

@@ -54,4 +54,79 @@ final List<User> foundUsers = couchbaseTemplate
   .consistentWith(QueryScanConsistency.REQUEST_PLUS)
   .all();
 ----
+====
+
+
+[[template.sub-document-ops]]
+== Sub-Document Operations
+
+Couchbase supports https://docs.couchbase.com/java-sdk/current/howtos/subdocument-operations.html[Sub-Document Operations]. This section documents how to use it with Spring Data Couchbase.
+
+
+
+Sub-Document operations may be quicker and more network-efficient than full-document operations such as upsert or replace because they only transmit the accessed sections of the document over the network.
+
+Sub-Document operations are also atomic, in that if one Sub-Document mutation fails then all will, allowing safe modifications to documents with built-in concurrency control.
+
+Currently Spring Data Couchbase supports only sub document mutations (remove, upsert, replace and insert).
+
+Mutation operations modify one or more paths in the document. The simplest of these operations is upsert, which, similar to the fulldoc-level upsert, will either modify the value of an existing path or create it if it does not exist:
+
+Following example will upsert the city field on the address of the user, without trasfering any additional user document data.
+
+.MutateIn upsert on the template
+====
+[source,java]
+----
+User user = new User();
+// id field on the base document id required
+user.setId(ID);
+user.setAddress(address);
+couchbaseTemplate.mutateInById(User.class)
+    .withUpsertPaths("address.city")
+    .one(user);
+----
+====
+
+[[template.sub-document-ops]]
+=== Executing Multiple Sub-Document Operations
+
+Multiple Sub-Document operations can be executed at once on the same document, allowing you to modify several Sub-Documents at once. When multiple operations are submitted within the context of a single mutateIn command, the server will execute all the operations with the same version of the document.
+
+To execute several mutation operations the method chaining can be used.
+
+.MutateIn Multiple Operations
+====
+[source,java]
+----
+couchbaseTemplate.mutateInById(User.class)
+    .withInsertPaths("roles", "subuser.firstname")
+    .withRemovePaths("address.city")
+    .withUpsertPaths("firstname")
+    .withReplacePaths("address.street")
+    .one(user);
+----
+====
+
+[[template.sub-document-cas]]
+=== Concurrent Modifications
+
+Concurrent Sub-Document operations on different parts of a document will not conflict so by default the CAS value will be not be supplied when executing the mutations.
+If CAS is required then it can be provided like this:
+
+.MutateIn With CAS
+====
+[source,java]
+----
+User user = new User();
+// id field on the base document id required
+user.setId(ID);
+// @Version field should have a value for CAS to be supplied
+user.setVersion(cas);
+user.setAddress(address);
+couchbaseTemplate.mutateInById(User.class)
+    .withUpsertPaths("address.city")
+    .withCasProvided()
+    .one(user);
+----
 ====

src/main/java/org/springframework/data/couchbase/core/CouchbaseExceptionTranslator.java

@@ -19,6 +19,7 @@ package org.springframework.data.couchbase.core;
 import java.util.ConcurrentModificationException;
 import java.util.concurrent.TimeoutException;
 
+import com.couchbase.client.core.error.subdoc.*;
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.DataAccessResourceFailureException;
 import org.springframework.dao.DataIntegrityViolationException;
@@ -63,6 +64,7 @@ import com.couchbase.client.core.error.transaction.TransactionOperationFailedExc
  * @author Simon Baslé
  * @author Michael Reiche
  * @author Graham Pople
+ * @author Tigran Babloyan
  */
 public class CouchbaseExceptionTranslator implements PersistenceExceptionTranslator {
 
@@ -102,7 +104,11 @@ public class CouchbaseExceptionTranslator implements PersistenceExceptionTransla
 			return new OperationCancellationException(ex.getMessage(), ex);
 		}
 
-		if (ex instanceof DesignDocumentNotFoundException || ex instanceof ValueTooLargeException) {
+		if (ex instanceof DesignDocumentNotFoundException || ex instanceof ValueTooLargeException
+				|| ex instanceof PathExistsException || ex instanceof PathInvalidException
+				|| ex instanceof PathNotFoundException || ex instanceof PathMismatchException
+				|| ex instanceof PathTooDeepException || ex instanceof ValueInvalidException
+				|| ex instanceof ValueTooDeepException || ex instanceof DocumentTooDeepException) {
 			return new InvalidDataAccessResourceUsageException(ex.getMessage(), ex);
 		}
 

src/main/java/org/springframework/data/couchbase/core/CouchbaseTemplate.java

@@ -41,6 +41,7 @@ import com.couchbase.client.java.query.QueryScanConsistency;
  * @author Michael Nitschinger
  * @author Michael Reiche
  * @author Jorge Rodriguez Martin
+ * @author Tigran Babloyan
  * @since 3.0
  */
 public class CouchbaseTemplate implements CouchbaseOperations, ApplicationContextAware {
@@ -95,6 +96,11 @@ public class CouchbaseTemplate implements CouchbaseOperations, ApplicationContex
 		return new ExecutableUpsertByIdOperationSupport(this).upsertById(domainType);
 	}
 
+	@Override
+	public <T> ExecutableMutateInById<T> mutateInById(Class<T> domainType) {
+		return new ExecutableMutateInByIdOperationSupport(this).mutateInById(domainType);
+	}
+
 	@Override
 	public <T> ExecutableInsertById<T> insertById(Class<T> domainType) {
 		return new ExecutableInsertByIdOperationSupport(this).insertById(domainType);

src/main/java/org/springframework/data/couchbase/core/ExecutableMutateInByIdOperation.java

@@ -0,0 +1,169 @@
+/*
+ * Copyright 2012-2023 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.core;
+
+import com.couchbase.client.core.msg.kv.DurabilityLevel;
+import com.couchbase.client.java.kv.MutateInOptions;
+import com.couchbase.client.java.kv.PersistTo;
+import com.couchbase.client.java.kv.ReplicateTo;
+import org.springframework.data.couchbase.core.support.*;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.time.Duration;
+import java.util.Collection;
+
+/**
+ * Mutate In Operations
+ *
+ * @author Tigran Babloyan
+ * @since 5.1
+ */
+public interface ExecutableMutateInByIdOperation {
+
+	/**
+	 * Mutate using the KV service.
+	 *
+	 * @param domainType the entity type to mutate.
+	 */
+	<T> ExecutableMutateInById<T> mutateInById(Class<T> domainType);
+
+	/**
+	 * Terminating operations invoking the actual execution.
+	 */
+	interface TerminatingMutateInById<T> extends OneAndAllEntity<T> {
+
+		/**
+		 * Insert one entity.
+		 *
+		 * @return Inserted entity.
+		 */
+		@Override
+		T one(T object);
+
+		/**
+		 * Insert a collection of entities.
+		 *
+		 * @return Inserted entities
+		 */
+		@Override
+		Collection<? extends T> all(Collection<? extends T> objects);
+
+	}
+
+	interface MutateInByIdWithPaths<T> extends TerminatingMutateInById<T>, WithMutateInPaths<T> {
+		/**
+         * Adds given paths to remove mutations.
+		 * See {@link com.couchbase.client.java.kv.Remove} for more details.
+		 * @param removePaths The property paths to removed from document.
+		 */
+		@Override
+		MutateInByIdWithPaths<T> withRemovePaths(final String... removePaths);
+		/**
+		 * Adds given paths to insert mutations.
+		 * See {@link com.couchbase.client.java.kv.Insert} for more details.
+		 * @param insertPaths The property paths to be inserted into the document.
+		 */
+		@Override
+		MutateInByIdWithPaths<T> withInsertPaths(final String... insertPaths);
+		/**
+		 * Adds given paths to upsert mutations.
+		 * See {@link com.couchbase.client.java.kv.Upsert} for more details.
+		 * @param upsertPaths The property paths to be upserted into the document.
+		 */
+		@Override
+		MutateInByIdWithPaths<T> withUpsertPaths(final String... upsertPaths);
+		/**
+		 * Adds given paths to replace mutations.
+		 * See {@link com.couchbase.client.java.kv.Replace} for more details.
+		 * @param replacePaths The property paths to be replaced in the document.
+		 */
+		@Override
+		MutateInByIdWithPaths<T> withReplacePaths(final String... replacePaths);
+		/**
+		 * Marks that the CAS value should be provided with the mutations to protect against concurrent modifications.
+		 * By default the CAS value is not provided.
+		 */
+		MutateInByIdWithPaths<T> withCasProvided();
+	}
+
+	/**
+	 * Fluent method to specify options.
+	 *
+	 * @param <T> the entity type to use.
+	 */
+	interface MutateInByIdWithOptions<T> extends MutateInByIdWithPaths<T>, WithMutateInOptions<T> {
+		/**
+		 * Fluent method to specify options to use for execution
+		 *
+		 * @param options to use for execution
+		 */
+		@Override
+		TerminatingMutateInById<T> withOptions(MutateInOptions options);
+	}
+
+	/**
+	 * Fluent method to specify the collection.
+	 *
+	 * @param <T> the entity type to use for the results.
+	 */
+	interface MutateInByIdInCollection<T> extends MutateInByIdWithOptions<T>, InCollection<Object> {
+		/**
+		 * With a different collection
+		 *
+		 * @param collection the collection to use.
+		 */
+		@Override
+		MutateInByIdWithOptions<T> inCollection(String collection);
+	}
+
+	/**
+	 * Fluent method to specify the scope.
+	 *
+	 * @param <T> the entity type to use for the results.
+	 */
+	interface MutateInByIdInScope<T> extends MutateInByIdInCollection<T>, InScope<Object> {
+		/**
+		 * With a different scope
+		 *
+		 * @param scope the scope to use.
+		 */
+		@Override
+		MutateInByIdInCollection<T> inScope(String scope);
+	}
+
+	interface MutateInByIdWithDurability<T> extends MutateInByIdInScope<T>, WithDurability<T> {
+		@Override
+		MutateInByIdInScope<T> withDurability(DurabilityLevel durabilityLevel);
+
+		@Override
+		MutateInByIdInScope<T> withDurability(PersistTo persistTo, ReplicateTo replicateTo);
+
+	}
+
+	interface MutateInByIdWithExpiry<T> extends MutateInByIdWithDurability<T>, WithExpiry<T> {
+		@Override
+		MutateInByIdWithDurability<T> withExpiry(Duration expiry);
+	}
+
+	/**
+	 * Provides methods for constructing KV operations in a fluent way.
+	 *
+	 * @param <T> the entity type to upsert
+	 */
+	interface ExecutableMutateInById<T> extends MutateInByIdWithExpiry<T> {}
+
+}

src/main/java/org/springframework/data/couchbase/core/ExecutableMutateInByIdOperationSupport.java

@@ -0,0 +1,188 @@
+/*
+ * Copyright 2012-2023 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.core;
+
+import com.couchbase.client.core.msg.kv.DurabilityLevel;
+import com.couchbase.client.java.kv.MutateInOptions;
+import com.couchbase.client.java.kv.MutateInSpec;
+import com.couchbase.client.java.kv.PersistTo;
+import com.couchbase.client.java.kv.ReplicateTo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.couchbase.core.mapping.CouchbaseDocument;
+import org.springframework.data.couchbase.core.query.OptionsBuilder;
+import org.springframework.data.couchbase.core.support.PseudoArgs;
+import org.springframework.util.Assert;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.time.Duration;
+import java.util.*;
+
+/**
+ * {@link ExecutableMutateInByIdOperation} implementations for Couchbase.
+ *
+ * @author Tigran Babloyan
+ */
+public class ExecutableMutateInByIdOperationSupport implements ExecutableMutateInByIdOperation {
+
+	private final CouchbaseTemplate template;
+	private static final Logger LOG = LoggerFactory.getLogger(ExecutableMutateInByIdOperationSupport.class);
+
+	public ExecutableMutateInByIdOperationSupport(final CouchbaseTemplate template) {
+		this.template = template;
+	}
+
+	@Override
+	public <T> ExecutableMutateInById<T> mutateInById(final Class<T> domainType) {
+		Assert.notNull(domainType, "DomainType must not be null!");
+		return new ExecutableMutateInByIdSupport(template, domainType, OptionsBuilder.getScopeFrom(domainType),
+				OptionsBuilder.getCollectionFrom(domainType), null, OptionsBuilder.getPersistTo(domainType),
+				OptionsBuilder.getReplicateTo(domainType), OptionsBuilder.getDurabilityLevel(domainType),
+				null, Collections.emptyList(), Collections.emptyList(), Collections.emptyList(),
+				Collections.emptyList(), false);
+	}
+
+	static class ExecutableMutateInByIdSupport<T> implements ExecutableMutateInById<T> {
+
+		private final CouchbaseTemplate template;
+		private final Class<T> domainType;
+		private final String scope;
+		private final String collection;
+		private final MutateInOptions options;
+		private final PersistTo persistTo;
+		private final ReplicateTo replicateTo;
+		private final DurabilityLevel durabilityLevel;
+		private final Duration expiry;
+		private final boolean provideCas;
+		private final List<String> removePaths = new ArrayList<>();
+		private final List<String> upsertPaths = new ArrayList<>();
+		private final List<String> insertPaths = new ArrayList<>();
+		private final List<String> replacePaths = new ArrayList<>();
+		private final ReactiveMutateInByIdOperationSupport.ReactiveMutateInByIdSupport<T> reactiveSupport;
+
+
+		ExecutableMutateInByIdSupport(final CouchbaseTemplate template, final Class<T> domainType, final String scope,
+				final String collection, final MutateInOptions options, final PersistTo persistTo, final ReplicateTo replicateTo,
+				final DurabilityLevel durabilityLevel, final Duration expiry, final List<String> removePaths, 
+				final List<String> upsertPaths, final List<String> insertPaths, final List<String> replacePaths, final boolean provideCas) {
+			this.template = template;
+			this.domainType = domainType;
+			this.scope = scope;
+			this.collection = collection;
+			this.options = options;
+			this.persistTo = persistTo;
+			this.replicateTo = replicateTo;
+			this.durabilityLevel = durabilityLevel;
+			this.expiry = expiry;
+			this.removePaths.addAll(removePaths);	
+			this.upsertPaths.addAll(upsertPaths);
+			this.insertPaths.addAll(insertPaths);
+			this.replacePaths.addAll(replacePaths);
+			this.provideCas = provideCas;
+			this.reactiveSupport = new ReactiveMutateInByIdOperationSupport.ReactiveMutateInByIdSupport<T>(template.reactive(), domainType, scope, collection,
+					options, persistTo, replicateTo, durabilityLevel, expiry, new NonReactiveSupportWrapper(template.support()), removePaths, upsertPaths, insertPaths, replacePaths, provideCas);
+		}
+
+		@Override
+		public T one(final T object) {
+			return reactiveSupport.one(object).block();
+		}
+
+		@Override
+		public Collection<? extends T> all(Collection<? extends T> objects) {
+			return reactiveSupport.all(objects).collectList().block();
+		}
+
+		@Override
+		public TerminatingMutateInById<T> withOptions(final MutateInOptions options) {
+			Assert.notNull(options, "Options must not be null.");
+			return new ExecutableMutateInByIdSupport(template, domainType, scope, collection, options, persistTo, replicateTo,
+					durabilityLevel, expiry, removePaths, upsertPaths, insertPaths, replacePaths, provideCas);
+		}
+
+		@Override
+		public MutateInByIdWithDurability<T> inCollection(final String collection) {
+			return new ExecutableMutateInByIdSupport<>(template, domainType, scope,
+					collection != null ? collection : this.collection, options, persistTo, replicateTo, durabilityLevel, expiry,
+					removePaths, upsertPaths, insertPaths, replacePaths, provideCas);
+		}
+
+		@Override
+		public MutateInByIdInCollection<T> inScope(final String scope) {
+			return new ExecutableMutateInByIdSupport<>(template, domainType, scope != null ? scope : this.scope, collection,
+					options, persistTo, replicateTo, durabilityLevel, expiry, removePaths, upsertPaths, insertPaths,
+					replacePaths, provideCas);
+		}
+
+		@Override
+		public MutateInByIdInScope<T> withDurability(final DurabilityLevel durabilityLevel) {
+			Assert.notNull(durabilityLevel, "Durability Level must not be null.");
+			return new ExecutableMutateInByIdSupport<>(template, domainType, scope, collection, options, persistTo, replicateTo,
+					durabilityLevel, expiry, removePaths, upsertPaths, insertPaths, replacePaths, provideCas);
+		}
+
+		@Override
+		public MutateInByIdInScope<T> withDurability(final PersistTo persistTo, final ReplicateTo replicateTo) {
+			Assert.notNull(persistTo, "PersistTo must not be null.");
+			Assert.notNull(replicateTo, "ReplicateTo must not be null.");
+			return new ExecutableMutateInByIdSupport<>(template, domainType, scope, collection, options, persistTo, replicateTo,
+					durabilityLevel, expiry, removePaths, upsertPaths, insertPaths, replacePaths, provideCas);
+		}
+
+		@Override
+		public MutateInByIdWithDurability<T> withExpiry(final Duration expiry) {
+			Assert.notNull(expiry, "expiry must not be null.");
+			return new ExecutableMutateInByIdSupport<>(template, domainType, scope, collection, options, persistTo, replicateTo,
+					durabilityLevel, expiry, removePaths, upsertPaths, insertPaths, replacePaths, provideCas);
+		}
+
+		@Override
+		public MutateInByIdWithDurability<T> withRemovePaths(final String... removePaths) {
+			Assert.notNull(removePaths, "removePaths path must not be null.");
+			return new ExecutableMutateInByIdSupport<>(template, domainType, scope, collection, options, persistTo, replicateTo,
+					durabilityLevel, expiry, Arrays.asList(removePaths), upsertPaths, insertPaths, replacePaths, provideCas);
+		}
+
+		@Override
+		public MutateInByIdWithDurability<T> withUpsertPaths(final String... upsertPaths) {
+			Assert.notNull(upsertPaths, "upsertPaths path must not be null.");
+			return new ExecutableMutateInByIdSupport<>(template, domainType, scope, collection, options, persistTo, replicateTo,
+					durabilityLevel, expiry, removePaths, Arrays.asList(upsertPaths), insertPaths, replacePaths, provideCas);
+		}
+
+		@Override
+		public MutateInByIdWithDurability<T> withInsertPaths(final String... insertPaths) {
+			Assert.notNull(insertPaths, "insertPaths path must not be null.");
+			return new ExecutableMutateInByIdSupport<>(template, domainType, scope, collection, options, persistTo, replicateTo,
+					durabilityLevel, expiry, removePaths, upsertPaths, Arrays.asList(insertPaths), replacePaths, provideCas);
+		}
+
+		@Override
+		public MutateInByIdWithDurability<T> withReplacePaths(final String... replacePaths) {
+			Assert.notNull(replacePaths, "replacePaths path must not be null.");
+			return new ExecutableMutateInByIdSupport<>(template, domainType, scope, collection, options, persistTo, replicateTo,
+					durabilityLevel, expiry, removePaths, upsertPaths, insertPaths, Arrays.asList(replacePaths), provideCas);
+		}
+
+		@Override
+		public MutateInByIdWithPaths<T> withCasProvided() {
+			return new ExecutableMutateInByIdSupport<>(template, domainType, scope, collection, options, persistTo, replicateTo,
+					durabilityLevel, expiry, removePaths, upsertPaths, insertPaths, replacePaths, true);
+		}
+	}
+
+}

src/main/java/org/springframework/data/couchbase/core/FluentCouchbaseOperations.java

@@ -22,4 +22,4 @@ package org.springframework.data.couchbase.core;
 public interface FluentCouchbaseOperations extends ExecutableUpsertByIdOperation, ExecutableInsertByIdOperation,
 		ExecutableReplaceByIdOperation, ExecutableFindByIdOperation, ExecutableFindFromReplicasByIdOperation,
 		ExecutableFindByQueryOperation, ExecutableFindByAnalyticsOperation, ExecutableExistsByIdOperation,
-		ExecutableRemoveByIdOperation, ExecutableRemoveByQueryOperation {}
+		ExecutableRemoveByIdOperation, ExecutableRemoveByQueryOperation, ExecutableMutateInByIdOperation {}

src/main/java/org/springframework/data/couchbase/core/ReactiveCouchbaseTemplate.java

@@ -44,6 +44,7 @@ import com.couchbase.client.java.query.QueryScanConsistency;
  * @author Michael Reiche
  * @author Jorge Rodriguez Martin
  * @author Carlos Espinaco
+ * @author Tigran Babloyan
  */
 public class ReactiveCouchbaseTemplate implements ReactiveCouchbaseOperations, ApplicationContextAware {
 
@@ -183,6 +184,11 @@ public class ReactiveCouchbaseTemplate implements ReactiveCouchbaseOperations, A
 		return new ReactiveUpsertByIdOperationSupport(this).upsertById(domainType);
 	}
 
+	@Override
+	public <T> ReactiveMutateInById<T> mutateInById(Class<T> domainType) {
+		return new ReactiveMutateInByIdOperationSupport(this).mutateInById(domainType);
+	}
+
 	@Override
 	public String getBucketName() {
 		return clientFactory.getBucket().name();

src/main/java/org/springframework/data/couchbase/core/ReactiveFluentCouchbaseOperations.java

@@ -22,4 +22,4 @@ package org.springframework.data.couchbase.core;
 public interface ReactiveFluentCouchbaseOperations extends ReactiveUpsertByIdOperation, ReactiveInsertByIdOperation,
 		ReactiveReplaceByIdOperation, ReactiveFindByIdOperation, ReactiveExistsByIdOperation,
 		ReactiveFindByAnalyticsOperation, ReactiveFindFromReplicasByIdOperation, ReactiveFindByQueryOperation,
-		ReactiveRemoveByIdOperation, ReactiveRemoveByQueryOperation {}
+		ReactiveRemoveByIdOperation, ReactiveRemoveByQueryOperation, ReactiveMutateInByIdOperation {}

src/main/java/org/springframework/data/couchbase/core/ReactiveMutateInByIdOperation.java

@@ -0,0 +1,169 @@
+/*
+ * Copyright 2012-2023 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.core;
+
+import com.couchbase.client.core.msg.kv.DurabilityLevel;
+import com.couchbase.client.java.kv.MutateInOptions;
+import com.couchbase.client.java.kv.PersistTo;
+import com.couchbase.client.java.kv.ReplicateTo;
+import org.springframework.data.couchbase.core.support.*;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.time.Duration;
+import java.util.Collection;
+
+/**
+ * Mutate In Operations
+ *
+ * @author Tigran Babloyan
+ * @since 5.1
+ */
+public interface ReactiveMutateInByIdOperation {
+
+	/**
+	 * Mutate using the KV service.
+	 *
+	 * @param domainType the entity type to mutate.
+	 */
+	<T> ReactiveMutateInById<T> mutateInById(Class<T> domainType);
+
+	/**
+	 * Terminating operations invoking the actual execution.
+	 */
+	interface TerminatingMutateInById<T> extends OneAndAllEntityReactive<T> {
+
+		/**
+		 * Mutate one entity.
+		 *
+		 * @return Upserted entity.
+		 */
+		@Override
+		Mono<T> one(T object);
+
+		/**
+		 * Mutate a collection of entities.
+		 *
+		 * @return Inserted entities
+		 */
+		@Override
+		Flux<? extends T> all(Collection<? extends T> objects);
+
+	}
+
+	interface MutateInByIdWithPaths<T> extends TerminatingMutateInById<T>, WithMutateInPaths<T> {
+		/**
+		 * Adds given paths to remove mutations.
+		 * See {@link com.couchbase.client.java.kv.Remove} for more details.
+		 * @param removePaths The property paths to removed from document.
+		 */
+		@Override
+		MutateInByIdWithPaths<T> withRemovePaths(final String... removePaths);
+		/**
+		 * Adds given paths to insert mutations.
+		 * See {@link com.couchbase.client.java.kv.Insert} for more details.
+		 * @param insertPaths The property paths to be inserted into the document.
+		 */
+		@Override
+		MutateInByIdWithPaths<T> withInsertPaths(final String... insertPaths);
+		/**
+		 * Adds given paths to upsert mutations.
+		 * See {@link com.couchbase.client.java.kv.Upsert} for more details.
+		 * @param upsertPaths The property paths to be upserted into the document.
+		 */
+		@Override
+		MutateInByIdWithPaths<T> withUpsertPaths(final String... upsertPaths);
+		/**
+		 * Adds given paths to replace mutations.
+		 * See {@link com.couchbase.client.java.kv.Replace} for more details.
+		 * @param replacePaths The property paths to be replaced in the document.
+		 */
+		@Override
+		MutateInByIdWithPaths<T> withReplacePaths(final String... replacePaths);
+		/**
+		 * Marks that the CAS value should be provided with the mutations to protect against concurrent modifications.
+		 * By default the CAS value is not provided.
+		 */
+		MutateInByIdWithPaths<T> withCasProvided();
+	}
+
+	/**
+	 * Fluent method to specify options.
+	 *
+	 * @param <T> the entity type to use.
+	 */
+	interface MutateInByIdWithOptions<T> extends MutateInByIdWithPaths<T>, WithMutateInOptions<T> {
+		/**
+		 * Fluent method to specify options to use for execution
+		 *
+		 * @param options to use for execution
+		 */
+		@Override
+		TerminatingMutateInById<T> withOptions(MutateInOptions options);
+	}
+
+	/**
+	 * Fluent method to specify the collection.
+	 *
+	 * @param <T> the entity type to use for the results.
+	 */
+	interface MutateInByIdInCollection<T> extends MutateInByIdWithOptions<T>, InCollection<Object> {
+		/**
+		 * With a different collection
+		 *
+		 * @param collection the collection to use.
+		 */
+		@Override
+		MutateInByIdWithOptions<T> inCollection(String collection);
+	}
+
+	/**
+	 * Fluent method to specify the scope.
+	 *
+	 * @param <T> the entity type to use for the results.
+	 */
+	interface MutateInByIdInScope<T> extends MutateInByIdInCollection<T>, InScope<Object> {
+		/**
+		 * With a different scope
+		 *
+		 * @param scope the scope to use.
+		 */
+		@Override
+		MutateInByIdInCollection<T> inScope(String scope);
+	}
+
+	interface MutateInByIdWithDurability<T> extends MutateInByIdInScope<T>, WithDurability<T> {
+		@Override
+		MutateInByIdInScope<T> withDurability(DurabilityLevel durabilityLevel);
+
+		@Override
+		MutateInByIdInScope<T> withDurability(PersistTo persistTo, ReplicateTo replicateTo);
+
+	}
+
+	interface MutateInByIdWithExpiry<T> extends MutateInByIdWithDurability<T>, WithExpiry<T> {
+		@Override
+		MutateInByIdWithDurability<T> withExpiry(Duration expiry);
+	}
+
+	/**
+	 * Provides methods for constructing KV operations in a fluent way.
+	 *
+	 * @param <T> the entity type to upsert
+	 */
+	interface ReactiveMutateInById<T> extends MutateInByIdWithExpiry<T> {}
+
+}

src/main/java/org/springframework/data/couchbase/core/ReactiveMutateInByIdOperationSupport.java

@@ -0,0 +1,244 @@
+/*
+ * Copyright 2012-2023 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.core;
+
+import com.couchbase.client.core.msg.kv.DurabilityLevel;
+import com.couchbase.client.java.kv.MutateInOptions;
+import com.couchbase.client.java.kv.MutateInSpec;
+import com.couchbase.client.java.kv.PersistTo;
+import com.couchbase.client.java.kv.ReplicateTo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.couchbase.core.mapping.CouchbaseDocument;
+import org.springframework.data.couchbase.core.mapping.CouchbaseList;
+import org.springframework.data.couchbase.core.query.OptionsBuilder;
+import org.springframework.data.couchbase.core.support.PseudoArgs;
+import org.springframework.util.Assert;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.time.Duration;
+import java.util.*;
+
+/**
+ * {@link ReactiveMutateInByIdOperation} implementations for Couchbase.
+ *
+ * @author Tigran Babloyan
+ */
+public class ReactiveMutateInByIdOperationSupport implements ReactiveMutateInByIdOperation {
+
+	private final ReactiveCouchbaseTemplate template;
+	private static final Logger LOG = LoggerFactory.getLogger(ReactiveMutateInByIdOperationSupport.class);
+
+	public ReactiveMutateInByIdOperationSupport(final ReactiveCouchbaseTemplate template) {
+		this.template = template;
+	}
+
+	@Override
+	public <T> ReactiveMutateInById<T> mutateInById(final Class<T> domainType) {
+		Assert.notNull(domainType, "DomainType must not be null!");
+		return new ReactiveMutateInByIdSupport<>(template, domainType, OptionsBuilder.getScopeFrom(domainType),
+				OptionsBuilder.getCollectionFrom(domainType), null, OptionsBuilder.getPersistTo(domainType),
+				OptionsBuilder.getReplicateTo(domainType), OptionsBuilder.getDurabilityLevel(domainType),
+				null, template.support(), Collections.emptyList(), Collections.emptyList(), Collections.emptyList(),
+				Collections.emptyList(), false);
+	}
+
+	static class ReactiveMutateInByIdSupport<T> implements ReactiveMutateInById<T> {
+
+		private final ReactiveCouchbaseTemplate template;
+		private final Class<T> domainType;
+		private final String scope;
+		private final String collection;
+		private final MutateInOptions options;
+		private final PersistTo persistTo;
+		private final ReplicateTo replicateTo;
+		private final DurabilityLevel durabilityLevel;
+		private final Duration expiry;
+		private final ReactiveTemplateSupport support;
+		private final boolean provideCas;
+		private final List<String> removePaths = new ArrayList<>();
+		private final List<String> upsertPaths = new ArrayList<>();
+		private final List<String> insertPaths = new ArrayList<>();
+		private final List<String> replacePaths = new ArrayList<>();
+		
+
+		ReactiveMutateInByIdSupport(final ReactiveCouchbaseTemplate template, final Class<T> domainType, final String scope,
+				final String collection, final MutateInOptions options, final PersistTo persistTo, final ReplicateTo replicateTo,
+				final DurabilityLevel durabilityLevel, final Duration expiry, ReactiveTemplateSupport support, final List<String> removePaths, 
+				final List<String> upsertPaths, final List<String> insertPaths, final List<String> replacePaths, final boolean provideCas) {
+			this.template = template;
+			this.domainType = domainType;
+			this.scope = scope;
+			this.collection = collection;
+			this.options = options;
+			this.persistTo = persistTo;
+			this.replicateTo = replicateTo;
+			this.durabilityLevel = durabilityLevel;
+			this.expiry = expiry;
+			this.support = support;
+			this.removePaths.addAll(removePaths);	
+			this.upsertPaths.addAll(upsertPaths);
+			this.insertPaths.addAll(insertPaths);
+			this.replacePaths.addAll(replacePaths);
+			this.provideCas = provideCas;
+		}
+
+		@Override
+		public Mono<T> one(T object) {
+			PseudoArgs<MutateInOptions> pArgs = new PseudoArgs(template, scope, collection, options, domainType);
+			if (LOG.isDebugEnabled()) {
+				LOG.debug("upsertById object={} {}", object, pArgs);
+			}
+			
+			Mono<T> reactiveEntity = TransactionalSupport.verifyNotInTransaction("mutateInById")
+					.then(support.encodeEntity(object)).flatMap(converted -> {
+						return Mono
+								.just(template.getCouchbaseClientFactory().withScope(pArgs.getScope())
+										.getCollection(pArgs.getCollection()))
+								.flatMap(collection -> collection.reactive()
+										.mutateIn(converted.getId().toString(), getMutations(converted), buildMutateInOptions(pArgs.getOptions(), object, converted))
+										.flatMap(
+												result -> support.applyResult(object, converted, converted.getId(), result.cas(), null, null)));
+					});
+
+			return reactiveEntity.onErrorMap(throwable -> {
+				if (throwable instanceof RuntimeException) {
+					return template.potentiallyConvertRuntimeException((RuntimeException) throwable);
+				} else {
+					return throwable;
+				}
+			});
+		}
+
+		@Override
+		public Flux<? extends T> all(Collection<? extends T> objects) {
+			return Flux.fromIterable(objects).flatMap(this::one);
+		}
+
+		@Override
+		public TerminatingMutateInById<T> withOptions(final MutateInOptions options) {
+			Assert.notNull(options, "Options must not be null.");
+			return new ReactiveMutateInByIdSupport(template, domainType, scope, collection, options, persistTo, replicateTo,
+					durabilityLevel, expiry, support, removePaths, upsertPaths, insertPaths, replacePaths, provideCas);
+		}
+
+		@Override
+		public MutateInByIdWithDurability<T> inCollection(final String collection) {
+			return new ReactiveMutateInByIdSupport<>(template, domainType, scope,
+					collection != null ? collection : this.collection, options, persistTo, replicateTo, durabilityLevel, expiry,
+					support, removePaths, upsertPaths, insertPaths, replacePaths, provideCas);
+		}
+
+		@Override
+		public MutateInByIdInCollection<T> inScope(final String scope) {
+			return new ReactiveMutateInByIdSupport<>(template, domainType, scope != null ? scope : this.scope, collection,
+					options, persistTo, replicateTo, durabilityLevel, expiry, support, removePaths, upsertPaths, insertPaths,
+					replacePaths, provideCas);
+		}
+
+		@Override
+		public MutateInByIdInScope<T> withDurability(final DurabilityLevel durabilityLevel) {
+			Assert.notNull(durabilityLevel, "Durability Level must not be null.");
+			return new ReactiveMutateInByIdSupport<>(template, domainType, scope, collection, options, persistTo, replicateTo,
+					durabilityLevel, expiry, support, removePaths, upsertPaths, insertPaths, replacePaths, provideCas);
+		}
+
+		@Override
+		public MutateInByIdInScope<T> withDurability(final PersistTo persistTo, final ReplicateTo replicateTo) {
+			Assert.notNull(persistTo, "PersistTo must not be null.");
+			Assert.notNull(replicateTo, "ReplicateTo must not be null.");
+			return new ReactiveMutateInByIdSupport<>(template, domainType, scope, collection, options, persistTo, replicateTo,
+					durabilityLevel, expiry, support, removePaths, upsertPaths, insertPaths, replacePaths, provideCas);
+		}
+
+		@Override
+		public MutateInByIdWithDurability<T> withExpiry(final Duration expiry) {
+			Assert.notNull(expiry, "expiry must not be null.");
+			return new ReactiveMutateInByIdSupport<>(template, domainType, scope, collection, options, persistTo, replicateTo,
+					durabilityLevel, expiry, support, removePaths, upsertPaths, insertPaths, replacePaths, provideCas);
+		}
+
+		@Override
+		public MutateInByIdWithDurability<T> withRemovePaths(final String... removePaths) {
+			Assert.notNull(removePaths, "removePaths path must not be null.");
+			return new ReactiveMutateInByIdSupport<>(template, domainType, scope, collection, options, persistTo, replicateTo,
+					durabilityLevel, expiry, support, Arrays.asList(removePaths), upsertPaths, insertPaths, replacePaths, provideCas);
+		}
+
+		@Override
+		public MutateInByIdWithDurability<T> withUpsertPaths(final String... upsertPaths) {
+			Assert.notNull(upsertPaths, "upsertPaths path must not be null.");
+			return new ReactiveMutateInByIdSupport<>(template, domainType, scope, collection, options, persistTo, replicateTo,
+					durabilityLevel, expiry, support, removePaths, Arrays.asList(upsertPaths), insertPaths, replacePaths, provideCas);
+		}
+
+		@Override
+		public MutateInByIdWithDurability<T> withInsertPaths(final String... insertPaths) {
+			Assert.notNull(insertPaths, "insertPaths path must not be null.");
+			return new ReactiveMutateInByIdSupport<>(template, domainType, scope, collection, options, persistTo, replicateTo,
+					durabilityLevel, expiry, support, removePaths, upsertPaths, Arrays.asList(insertPaths), replacePaths, provideCas);
+		}
+
+		@Override
+		public MutateInByIdWithDurability<T> withReplacePaths(final String... replacePaths) {
+			Assert.notNull(replacePaths, "replacePaths path must not be null.");
+			return new ReactiveMutateInByIdSupport<>(template, domainType, scope, collection, options, persistTo, replicateTo,
+					durabilityLevel, expiry, support, removePaths, upsertPaths, insertPaths, Arrays.asList(replacePaths), provideCas);
+		}
+
+		@Override
+		public MutateInByIdWithPaths<T> withCasProvided() {
+			return new ReactiveMutateInByIdSupport<>(template, domainType, scope, collection, options, persistTo, replicateTo,
+					durabilityLevel, expiry, support, removePaths, upsertPaths, insertPaths, replacePaths, true);
+		}
+
+		private MutateInOptions buildMutateInOptions(MutateInOptions options, T object, CouchbaseDocument doc) {
+			return OptionsBuilder.buildMutateInOptions(options, persistTo, replicateTo, durabilityLevel, expiry, doc,
+					provideCas ? support.getCas(object) : null);
+		}
+
+		private List<MutateInSpec> getMutations(CouchbaseDocument document) {
+			List<MutateInSpec> mutations = new ArrayList<>();
+			for (String path : removePaths) {
+				mutations.add(MutateInSpec.remove(path));
+			}
+			for (String path : upsertPaths) {
+				mutations.add(MutateInSpec.upsert(path, getCouchbaseContent(document, path)).createPath());
+			}
+			for (String path : insertPaths) {
+				mutations.add(MutateInSpec.insert(path, getCouchbaseContent(document, path)).createPath());
+			}
+			for (String path : replacePaths) {
+				mutations.add(MutateInSpec.replace(path, getCouchbaseContent(document, path)));
+			}
+			return mutations;
+		}
+
+		private Object getCouchbaseContent(CouchbaseDocument document, String path) {
+			Object result = document.export();
+			for(var node : path.split("\\.")) {
+				if(result instanceof Map map) {
+					result = map.get(node);
+				} else {
+					throw new IllegalArgumentException("Path " + path + " is not valid.");
+				}
+			}
+			return result;
+		}
+	}
+
+}

src/main/java/org/springframework/data/couchbase/core/query/OptionsBuilder.java

@@ -32,7 +32,6 @@ import org.slf4j.LoggerFactory;
 import org.springframework.core.annotation.AnnotatedElementUtils;
 import org.springframework.data.couchbase.core.mapping.CouchbaseDocument;
 import org.springframework.data.couchbase.core.mapping.Document;
-import org.springframework.data.couchbase.core.mapping.Expiry;
 import org.springframework.data.couchbase.repository.Collection;
 import org.springframework.data.couchbase.repository.ScanConsistency;
 import org.springframework.data.couchbase.repository.Scope;
@@ -46,6 +45,7 @@ import com.couchbase.client.java.json.JsonObject;
 import com.couchbase.client.java.kv.ExistsOptions;
 import com.couchbase.client.java.kv.InsertOptions;
 import com.couchbase.client.java.kv.PersistTo;
+import com.couchbase.client.java.kv.MutateInOptions;
 import com.couchbase.client.java.kv.RemoveOptions;
 import com.couchbase.client.java.kv.ReplaceOptions;
 import com.couchbase.client.java.kv.ReplicateTo;
@@ -163,6 +163,28 @@ public class OptionsBuilder {
 		return options;
 	}
 
+	public static MutateInOptions buildMutateInOptions(MutateInOptions options, PersistTo persistTo, ReplicateTo replicateTo,
+													 DurabilityLevel durabilityLevel, Duration expiry, CouchbaseDocument doc, Long cas) {
+		options = options != null ? options : MutateInOptions.mutateInOptions();
+		if (persistTo != PersistTo.NONE || replicateTo != ReplicateTo.NONE) {
+			options.durability(persistTo, replicateTo);
+		} else if (durabilityLevel != DurabilityLevel.NONE) {
+			options.durability(durabilityLevel);
+		}
+		if (expiry != null) {
+			options.expiry(expiry);
+		} else if (doc.getExpiration() != 0) {
+			options.expiry(Duration.ofSeconds(doc.getExpiration()));
+		}
+		if (cas != null) {
+			options.cas(cas);
+		}
+		if (LOG.isDebugEnabled()) {
+			LOG.debug("mutate in options: {}" + toString(options));
+		}
+		return options;
+	}
+
 	public static ReplaceOptions buildReplaceOptions(ReplaceOptions options, PersistTo persistTo, ReplicateTo replicateTo,
 			DurabilityLevel durabilityLevel, Duration expiry, Long cas, CouchbaseDocument doc) {
 		options = options != null ? options : ReplaceOptions.replaceOptions();
@@ -332,6 +354,22 @@ public class OptionsBuilder {
 		return s.toString();
 	}
 
+	static String toString(MutateInOptions o) {
+		StringBuilder s = new StringBuilder();
+		MutateInOptions.Built b = o.build();
+		s.append("{");
+		s.append("cas: " + b.cas());
+		s.append(", durabilityLevel: " + b.durabilityLevel());
+		s.append(", persistTo: " + b.persistTo());
+		s.append(", replicateTo: " + b.replicateTo());
+		s.append(", timeout: " + b.timeout());
+		s.append(", retryStrategy: " + b.retryStrategy());
+		s.append(", clientContext: " + b.clientContext());
+		s.append(", parentSpan: " + b.parentSpan());
+		s.append("}");
+		return s.toString();
+	}
+
 	private static JsonObject getQueryOpts(QueryOptions.Built optsBuilt) {
 		JsonObject jo = JsonObject.create();
 		optsBuilt.injectParams(jo);

src/main/java/org/springframework/data/couchbase/core/support/WithMutateInOptions.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2021-2023 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.core.support;
+
+import com.couchbase.client.java.kv.MutateInOptions;
+
+/**
+ * A common interface for all of Insert, Replace, Upsert mutations that take options.
+ *
+ * @author Tigran Babloyan
+ * @param <T> - the entity class
+ */
+public interface WithMutateInOptions<T> {
+	/**
+	 * Specify options
+	 * 
+	 * @param options The mutate options to use.
+	 */
+	Object withOptions(MutateInOptions options);
+}

src/main/java/org/springframework/data/couchbase/core/support/WithMutateInPaths.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2021-2023 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.core.support;
+
+/**
+ * A common interface for all of Insert, Replace, Upsert and Remove mutations that take options.
+ *
+ * @author Tigran Babloyan
+ * @param <T> - the entity class
+ */
+public interface WithMutateInPaths<T> {
+	Object withRemovePaths(final String... removePaths);
+
+	Object withInsertPaths(final String... insertPaths);
+
+	Object withReplacePaths(final String... replacePaths);
+
+	Object withUpsertPaths(final String... upsertPaths);
+}