Fix 1441 in 4_4_x - scope and collection not considered in repo delete.

src/main/java/org/springframework/data/couchbase/repository/query/AbstractCouchbaseQuery.java

@@ -19,6 +19,7 @@
 import org.springframework.data.couchbase.core.CouchbaseOperations;
 import org.springframework.data.couchbase.core.ExecutableFindByQueryOperation.ExecutableFindByQuery;
 import org.springframework.data.couchbase.core.ExecutableFindByQueryOperation.TerminatingFindByQuery;
+import org.springframework.data.couchbase.core.ExecutableRemoveByQueryOperation.ExecutableRemoveByQuery;
 import org.springframework.data.couchbase.core.query.Query;
 import org.springframework.data.couchbase.repository.query.CouchbaseQueryExecution.DeleteExecution;
 import org.springframework.data.couchbase.repository.query.CouchbaseQueryExecution.PagedExecution;
@@ -43,6 +44,7 @@ public abstract class AbstractCouchbaseQuery extends AbstractCouchbaseQueryBase<
 		implements RepositoryQuery {
 
 	private final ExecutableFindByQuery<?> findOperationWithProjection;
+	private final ExecutableRemoveByQuery<?> removeOp;
 
 	/**
 	 * Creates a new {@link AbstractCouchbaseQuery} from the given {@link ReactiveCouchbaseQueryMethod} and
@@ -65,6 +67,8 @@ public AbstractCouchbaseQuery(CouchbaseQueryMethod method, CouchbaseOperations o
 		ExecutableFindByQuery<?> findOp = operations.findByQuery(type);
 		findOp = (ExecutableFindByQuery<?>) (findOp.inScope(method.getScope()).inCollection(method.getCollection()));
 		this.findOperationWithProjection = findOp;
+		this.removeOp = (ExecutableRemoveByQuery<?>) (operations.removeByQuery(type).inScope(method.getScope())
+				.inCollection(method.getCollection()));
 	}
 
 	/**
@@ -114,7 +118,7 @@ private CouchbaseQueryExecution getExecution(ParameterAccessor accessor, Convert
 	private CouchbaseQueryExecution getExecutionToWrap(ParameterAccessor accessor, ExecutableFindByQuery<?> operation) {
 
 		if (isDeleteQuery()) {
-			return new DeleteExecution(getOperations(), getQueryMethod());
+			return new DeleteExecution(removeOp);
 		} else if (isTailable(getQueryMethod())) {
 			return (q, t, r, c) -> operation.as(r).matching(q.with(accessor.getPageable())).all(); // s/b tail() instead of
 			// all()

src/main/java/org/springframework/data/couchbase/repository/query/AbstractReactiveCouchbaseQuery.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2021 the original author or authors.
+ * Copyright 2020-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.
@@ -19,6 +19,7 @@
 import org.springframework.data.couchbase.core.ReactiveCouchbaseOperations;
 import org.springframework.data.couchbase.core.ReactiveFindByQueryOperation;
 import org.springframework.data.couchbase.core.ReactiveFindByQueryOperation.ReactiveFindByQuery;
+import org.springframework.data.couchbase.core.ReactiveRemoveByQueryOperation.ReactiveRemoveByQuery;
 import org.springframework.data.couchbase.core.query.Query;
 import org.springframework.data.couchbase.repository.query.ReactiveCouchbaseQueryExecution.DeleteExecution;
 import org.springframework.data.couchbase.repository.query.ReactiveCouchbaseQueryExecution.ResultProcessingExecution;
@@ -41,7 +42,8 @@
 public abstract class AbstractReactiveCouchbaseQuery extends AbstractCouchbaseQueryBase<ReactiveCouchbaseOperations>
 		implements RepositoryQuery {
 
-	private final ReactiveFindByQuery<?> findOperationWithProjection;
+	private final ReactiveFindByQuery<?> findOp;
+	private final ReactiveRemoveByQuery<?> removeOp;
 
 	/**
 	 * Creates a new {@link AbstractReactiveCouchbaseQuery} from the given {@link ReactiveCouchbaseQueryMethod} and
@@ -62,9 +64,10 @@ public AbstractReactiveCouchbaseQuery(ReactiveCouchbaseQueryMethod method, React
 
 		EntityMetadata<?> metadata = method.getEntityInformation();
 		Class<?> type = metadata.getJavaType();
-		ReactiveFindByQuery<?> findOp = operations.findByQuery(type);
-		findOp = (ReactiveFindByQuery<?>) (findOp.inScope(method.getScope()).inCollection(method.getCollection()));
-		this.findOperationWithProjection = findOp;
+		this.findOp = (ReactiveFindByQuery<?>) (operations.findByQuery(type).inScope(method.getScope())
+				.inCollection(method.getCollection()));
+		this.removeOp = (ReactiveRemoveByQuery<?>) (operations.removeByQuery(type).inScope(method.getScope())
+				.inCollection(method.getCollection()));
 	}
 
 	/**
@@ -83,10 +86,8 @@ protected Object doExecute(CouchbaseQueryMethod method, ResultProcessor processo
 		// query = applyAnnotatedCollationIfPresent(query, accessor); // not yet implemented
 		query = applyQueryMetaAttributesIfPresent(query, typeToRead);
 
-		ReactiveFindByQuery<?> find = findOperationWithProjection;
-
 		ReactiveCouchbaseQueryExecution execution = getExecution(accessor,
-				new ResultProcessingConverter<>(processor, getOperations(), getInstantiators()), find);
+				new ResultProcessingConverter<>(processor, getOperations(), getInstantiators()), findOp);
 		return execution.execute(query, processor.getReturnedType().getDomainType(), typeToRead, null);
 	}
 
@@ -113,7 +114,7 @@ private ReactiveCouchbaseQueryExecution getExecutionToWrap(ParameterAccessor acc
 			ReactiveFindByQuery<?> operation) {
 
 		if (isDeleteQuery()) {
-			return new DeleteExecution(getOperations(), getQueryMethod());
+			return new DeleteExecution(removeOp);
 		} else if (isTailable(getQueryMethod())) {
 			return (q, t, r, c) -> operation.as(r).matching(q.with(accessor.getPageable())).all(); // s/b tail() instead of
 																																															// all()

src/main/java/org/springframework/data/couchbase/repository/query/CouchbaseQueryExecution.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-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.
@@ -18,15 +18,14 @@
 import java.util.List;
 
 import org.springframework.core.convert.converter.Converter;
-import org.springframework.data.couchbase.core.CouchbaseOperations;
 import org.springframework.data.couchbase.core.ExecutableFindByQueryOperation.ExecutableFindByQuery;
 import org.springframework.data.couchbase.core.ExecutableFindByQueryOperation.TerminatingFindByQuery;
+import org.springframework.data.couchbase.core.ExecutableRemoveByQueryOperation.ExecutableRemoveByQuery;
 import org.springframework.data.couchbase.core.query.Query;
 import org.springframework.data.domain.PageImpl;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Slice;
 import org.springframework.data.domain.SliceImpl;
-import org.springframework.data.repository.query.QueryMethod;
 import org.springframework.util.Assert;
 
 /**
@@ -47,12 +46,10 @@ interface CouchbaseQueryExecution {
 
 	final class DeleteExecution implements CouchbaseQueryExecution {
 
-		private final CouchbaseOperations operations;
-		private final QueryMethod method;
+		private final ExecutableRemoveByQuery<?> removeOperation;
 
-		public DeleteExecution(CouchbaseOperations operations, QueryMethod method) {
-			this.operations = operations;
-			this.method = method;
+		public DeleteExecution(ExecutableRemoveByQuery<?> removeOperation) {
+			this.removeOperation = removeOperation;
 		}
 
 		/*
@@ -61,7 +58,7 @@ public DeleteExecution(CouchbaseOperations operations, QueryMethod method) {
 		 */
 		@Override
 		public Object execute(Query query, Class<?> type, Class<?> returnType, String collection) {
-			return operations.removeByQuery(type).matching(query).all();
+			return removeOperation.matching(query).all();
 		}
 
 	}

src/main/java/org/springframework/data/couchbase/repository/query/ReactiveCouchbaseQueryExecution.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-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,7 +16,7 @@
 package org.springframework.data.couchbase.repository.query;
 
 import org.springframework.core.convert.converter.Converter;
-import org.springframework.data.couchbase.core.ReactiveCouchbaseOperations;
+import org.springframework.data.couchbase.core.ReactiveRemoveByQueryOperation;
 import org.springframework.data.couchbase.core.query.Query;
 import org.springframework.util.Assert;
 
@@ -39,12 +39,10 @@ interface ReactiveCouchbaseQueryExecution {
 
 	final class DeleteExecution implements ReactiveCouchbaseQueryExecution {
 
-		private final ReactiveCouchbaseOperations operations;
-		private final CouchbaseQueryMethod method;
+		private final ReactiveRemoveByQueryOperation.ReactiveRemoveByQuery removeOp;
 
-		public DeleteExecution(ReactiveCouchbaseOperations operations, CouchbaseQueryMethod method) {
-			this.operations = operations;
-			this.method = method;
+		public DeleteExecution(ReactiveRemoveByQueryOperation.ReactiveRemoveByQuery<?> removeOp) {
+			this.removeOp = removeOp;
 		}
 
 		/*
@@ -53,7 +51,7 @@ public DeleteExecution(ReactiveCouchbaseOperations operations, CouchbaseQueryMet
 		 */
 		@Override
 		public Object execute(Query query, Class<?> type, Class<?> returnType, String collection) {
-			return operations.removeByQuery(type)/*.inCollection(collection)*/.matching(query).all();
+			return removeOp.matching(query).all();
 		}
 
 	}

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

@@ -104,6 +104,11 @@ List<Airport> findByIataInAndIcaoIn(java.util.Collection<String> size, java.util
 	@ScanConsistency(query = QueryScanConsistency.REQUEST_PLUS)
 	List<RemoveResult> deleteByIata(String iata);
 
+	@Query("#{#n1ql.delete} WHERE #{#n1ql.filter} and  iata = $1 #{#n1ql.returning}")
+	@ScanConsistency(query = QueryScanConsistency.REQUEST_PLUS)
+	@Collection("bogus_collection")
+	List<RemoveResult> deleteByIataAnnotated(String iata);
+
 	@Query("SELECT __cas, * from #{#n1ql.bucket} where iata = $1")
 	@ScanConsistency(query = QueryScanConsistency.REQUEST_PLUS)
 	List<Airport> getAllByIataNoID(String iata);

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

@@ -0,0 +1,26 @@
+/*
+ * Copyright 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.
+ * 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.Collection;
+
+/**
+ * AirportRepository with collection annotation
+ *
+ * @author Michael Reiche
+ */
+@Collection("my_collection2")
+public interface AirportRepositoryAnnotated extends AirportRepository {}

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

@@ -105,7 +105,7 @@ public void beforeEach() {
 	static List<String> keyList = Arrays.asList("a", "b", "c", "d", "e");
 	static Collection collection;
 	static ReactiveCollection rCollection;
-	@Autowired ReactiveAirportRepository airportRepository; // intellij flags "Could not Autowire", but it runs ok.
+	@Autowired ReactiveAirportRepository reactiveAirportRepository; // intellij flags "Could not Autowire", runs ok
 
 	AtomicInteger rCat = new AtomicInteger(0);
 	AtomicInteger rFlat = new AtomicInteger(0);
@@ -136,7 +136,7 @@ public void cbse() {
 			listOfLists.add(list);
 		}
 		Flux<Object> af = Flux.fromIterable(listOfLists).concatMap(catalogToStore -> Flux.fromIterable(catalogToStore)
-				.parallel(4).runOn(Schedulers.parallel()).concatMap((entity) -> airportRepository.save(entity)));
+				.parallel(4).runOn(Schedulers.parallel()).concatMap((entity) -> reactiveAirportRepository.save(entity)));
 		List<Object> saved = af.collectList().block();
 		System.out.println("results.size() : " + saved.size());
 
@@ -152,7 +152,7 @@ public void cbse() {
 			e.printStackTrace();
 			throw e;
 		}
-		List<Airport> airports = airportRepository.findAll().collectList().block();
+		List<Airport> airports = reactiveAirportRepository.findAll().collectList().block();
 		assertEquals(0, airports.size(), "should have been all deleted");
 	}
 
@@ -164,11 +164,11 @@ public void pairIdAndResult() {
 		for (int i = 0; i < 5; i++) {
 			list.add(a.withId(UUID.randomUUID().toString()));
 		}
-		Flux<Object> af = Flux.fromIterable(list).concatMap((entity) -> airportRepository.save(entity));
+		Flux<Object> af = Flux.fromIterable(list).concatMap((entity) -> reactiveAirportRepository.save(entity));
 		List<Object> saved = af.collectList().block();
 		System.out.println("results.size() : " + saved.size());
 		Flux<Pair<String, Mono<Airport>>> pairFlux = Flux.fromIterable(list)
-				.map((airport) -> Pair.of(airport.getId(), airportRepository.findById(airport.getId())));
+				.map((airport) -> Pair.of(airport.getId(), reactiveAirportRepository.findById(airport.getId())));
 		List<Pair<String, Mono<Airport>>> airportPairs = pairFlux.collectList().block();
 		for (Pair<String, Mono<Airport>> airportPair : airportPairs) {
 			System.out.println("id: " + airportPair.getFirst() + " airport: " + airportPair.getSecond().block());

src/test/java/org/springframework/data/couchbase/domain/ReactiveAirportRepository.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.
@@ -21,6 +21,8 @@
 
 import java.util.ArrayList;
 
+import org.springframework.data.couchbase.core.RemoveResult;
+import org.springframework.data.couchbase.repository.Collection;
 import org.springframework.data.couchbase.repository.DynamicProxyable;
 import org.springframework.data.couchbase.repository.Query;
 import org.springframework.data.couchbase.repository.ReactiveCouchbaseRepository;
@@ -99,6 +101,15 @@ public interface ReactiveAirportRepository
 	@ScanConsistency(query = QueryScanConsistency.REQUEST_PLUS)
 	Mono<Airport> findByIata(String iata);
 
+	@Query("#{#n1ql.delete} WHERE #{#n1ql.filter} and iata = $1 #{#n1ql.returning}")
+	@ScanConsistency(query = QueryScanConsistency.REQUEST_PLUS)
+	Flux<RemoveResult> deleteByIata(String iata);
+
+	@Query("#{#n1ql.delete} WHERE #{#n1ql.filter} and iata = $1 #{#n1ql.returning}")
+	@ScanConsistency(query = QueryScanConsistency.REQUEST_PLUS)
+	@Collection("bogus_collection")
+	Flux<RemoveResult> deleteByIataAnnotated(String iata);
+
 	// This is not efficient. See findAllByIataLike for efficient reactive paging
 	default public Mono<Page<Airport>> findAllAirportsPaged(Pageable pageable) {
 		return count().flatMap(airportCount -> {

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

@@ -0,0 +1,26 @@
+/*
+ * Copyright 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.
+ * 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.Collection;
+
+/**
+ * AirportRepository with collection annotation
+ *
+ * @author Michael Reiche
+ */
+@Collection("my_collection2")
+public interface ReactiveAirportRepositoryAnnotated extends ReactiveAirportRepository{}

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

@@ -60,7 +60,7 @@ public class ReactiveCouchbaseRepositoryKeyValueIntegrationTests extends Cluster
 
 	@Autowired ReactiveUserRepository userRepository;
 
-	@Autowired ReactiveAirportRepository airportRepository;
+	@Autowired ReactiveAirportRepository reactiveAirportRepository;
 
 	@Autowired ReactiveAirlineRepository airlineRepository;
 
@@ -108,13 +108,13 @@ void findByIdAudited() {
 		Airport vie = null;
 		try {
 			vie = new Airport("airports::vie", "vie", "low2");
-			Airport saved = airportRepository.save(vie).block();
-			Airport airport1 = airportRepository.findById(saved.getId()).block();
+			Airport saved = reactiveAirportRepository.save(vie).block();
+			Airport airport1 = reactiveAirportRepository.findById(saved.getId()).block();
 			assertEquals(airport1, saved);
 			assertEquals(saved.getCreatedBy(), ReactiveNaiveAuditorAware.AUDITOR); // ReactiveNaiveAuditorAware will provide
 																																							// this
 		} finally {
-			airportRepository.delete(vie).block();
+			reactiveAirportRepository.delete(vie).block();
 		}
 	}