DATAMONGO-1870 - Consider skip/limit on MongoOperations.remove(Query, Class).

We now use _id lookup for remove operations that query with limit or skip parameters. This allows more fine grained control over documents removed.

Original pull request: #532.
Related pull request: #531.

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoOperations.java

@@ -769,8 +769,8 @@ public interface MongoOperations extends FluentMongoOperations {
 	}
 
 	/**
-	 * Triggers <a href="https://docs.mongodb.org/manual/reference/method/db.collection.findAndModify/">findAndModify<a/>
-	 * to apply provided {@link Update} on documents matching {@link Criteria} of given {@link Query}.
+	 * Triggers <a href="https://docs.mongodb.org/manual/reference/method/db.collection.findAndModify/">findAndModify
+	 * <a/>* to apply provided {@link Update} on documents matching {@link Criteria} of given {@link Query}.
 	 *
 	 * @param query the {@link Query} class that specifies the {@link Criteria} used to find a record and also an optional
 	 *          fields specification. Must not be {@literal null}.
@@ -782,8 +782,8 @@ public interface MongoOperations extends FluentMongoOperations {
 	<T> T findAndModify(Query query, Update update, Class<T> entityClass);
 
 	/**
-	 * Triggers <a href="https://docs.mongodb.org/manual/reference/method/db.collection.findAndModify/">findAndModify<a/>
-	 * to apply provided {@link Update} on documents matching {@link Criteria} of given {@link Query}.
+	 * Triggers <a href="https://docs.mongodb.org/manual/reference/method/db.collection.findAndModify/">findAndModify
+	 * <a/>* to apply provided {@link Update} on documents matching {@link Criteria} of given {@link Query}.
 	 *
 	 * @param query the {@link Query} class that specifies the {@link Criteria} used to find a record and also an optional
 	 *          fields specification. Must not be {@literal null}.
@@ -796,8 +796,8 @@ public interface MongoOperations extends FluentMongoOperations {
 	<T> T findAndModify(Query query, Update update, Class<T> entityClass, String collectionName);
 
 	/**
-	 * Triggers <a href="https://docs.mongodb.org/manual/reference/method/db.collection.findAndModify/">findAndModify<a/>
-	 * to apply provided {@link Update} on documents matching {@link Criteria} of given {@link Query} taking
+	 * Triggers <a href="https://docs.mongodb.org/manual/reference/method/db.collection.findAndModify/">findAndModify
+	 * <a/>* to apply provided {@link Update} on documents matching {@link Criteria} of given {@link Query} taking
 	 * {@link FindAndModifyOptions} into account.
 	 *
 	 * @param query the {@link Query} class that specifies the {@link Criteria} used to find a record and also an optional
@@ -813,8 +813,8 @@ public interface MongoOperations extends FluentMongoOperations {
 	<T> T findAndModify(Query query, Update update, FindAndModifyOptions options, Class<T> entityClass);
 
 	/**
-	 * Triggers <a href="https://docs.mongodb.org/manual/reference/method/db.collection.findAndModify/">findAndModify<a/>
-	 * to apply provided {@link Update} on documents matching {@link Criteria} of given {@link Query} taking
+	 * Triggers <a href="https://docs.mongodb.org/manual/reference/method/db.collection.findAndModify/">findAndModify
+	 * <a/>* to apply provided {@link Update} on documents matching {@link Criteria} of given {@link Query} taking
 	 * {@link FindAndModifyOptions} into account.
 	 *
 	 * @param query the {@link Query} class that specifies the {@link Criteria} used to find a record and also an optional
@@ -1142,6 +1142,7 @@ public interface MongoOperations extends FluentMongoOperations {
 	 * @param query the query document that specifies the criteria used to remove a record.
 	 * @param entityClass class that determines the collection to use.
 	 * @return the {@link DeleteResult} which lets you access the results of the previous delete.
+	 * @throws IllegalArgumentException when {@literal query} or {@literal entityClass} is {@literal null}.
 	 */
 	DeleteResult remove(Query query, Class<?> entityClass);
 
@@ -1153,6 +1154,8 @@ public interface MongoOperations extends FluentMongoOperations {
 	 * @param entityClass class of the pojo to be operated on. Can be {@literal null}.
 	 * @param collectionName name of the collection where the objects will removed, must not be {@literal null} or empty.
 	 * @return the {@link DeleteResult} which lets you access the results of the previous delete.
+	 * @throws IllegalArgumentException when {@literal query}, {@literal entityClass} or {@literal collectionName} is
+	 *           {@literal null}.
 	 */
 	DeleteResult remove(Query query, Class<?> entityClass, String collectionName);
 
@@ -1165,6 +1168,7 @@ public interface MongoOperations extends FluentMongoOperations {
 	 * @param query the query document that specifies the criteria used to remove a record.
 	 * @param collectionName name of the collection where the objects will removed, must not be {@literal null} or empty.
 	 * @return the {@link DeleteResult} which lets you access the results of the previous delete.
+	 * @throws IllegalArgumentException when {@literal query} or {@literal collectionName} is {@literal null}.
 	 */
 	DeleteResult remove(Query query, String collectionName);
 

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/MongoTemplate.java

@@ -24,19 +24,8 @@ import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.Map.Entry;
-import java.util.Optional;
-import java.util.Scanner;
-import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
@@ -135,6 +124,7 @@ import org.springframework.util.ObjectUtils;
 import org.springframework.util.ResourceUtils;
 import org.springframework.util.StringUtils;
 
+import com.mongodb.Cursor;
 import com.mongodb.DBCollection;
 import com.mongodb.DBCursor;
 import com.mongodb.Mongo;
@@ -1693,13 +1683,11 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 	protected <T> DeleteResult doRemove(final String collectionName, final Query query,
 			@Nullable final Class<T> entityClass) {
 
+		Assert.notNull(query, "Query must not be null!");
 		Assert.hasText(collectionName, "Collection name must not be null or empty!");
-		if (query == null) {
-			throw new InvalidDataAccessApiUsageException("Query passed in to remove can't be null!");
-		}
 
-		final Document queryObject = query.getQueryObject();
 		final MongoPersistentEntity<?> entity = getPersistentEntity(entityClass);
+		final Document queryObject = queryMapper.getMappedObject(query.getQueryObject(), entity);
 
 		return execute(collectionName, new CollectionCallback<DeleteResult>() {
 
@@ -1708,7 +1696,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 
 				maybeEmitEvent(new BeforeDeleteEvent<T>(queryObject, entityClass, collectionName));
 
-				Document mappedQuery = queryMapper.getMappedObject(queryObject, entity);
+				Document removeQuery = queryObject;
 
 				DeleteOptions options = new DeleteOptions();
 				query.getCollation().map(Collation::toMongoCollation).ifPresent(options::collation);
@@ -1721,13 +1709,26 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 				DeleteResult dr = null;
 				if (LOGGER.isDebugEnabled()) {
 					LOGGER.debug("Remove using query: {} in collection: {}.",
-							new Object[] { serializeToJsonSafely(mappedQuery), collectionName });
+							new Object[] { serializeToJsonSafely(removeQuery), collectionName });
+				}
+
+				if (query.getLimit() > 0 || query.getSkip() > 0) {
+
+					MongoCursor<Document> cursor = new QueryCursorPreparer(query, entityClass)
+							.prepare(collection.find(removeQuery).projection(new Document(ID_FIELD, 1))).iterator();
+
+					Set<Object> ids = new LinkedHashSet<>();
+					while (cursor.hasNext()) {
+						ids.add(cursor.next().get(ID_FIELD));
+					}
+
+					removeQuery = new Document(ID_FIELD, new Document("$in", ids));
 				}
 
 				if (writeConcernToUse == null) {
-					dr = collection.deleteMany(mappedQuery, options);
+					dr = collection.deleteMany(removeQuery, options);
 				} else {
-					dr = collection.withWriteConcern(writeConcernToUse).deleteMany(mappedQuery, options);
+					dr = collection.withWriteConcern(writeConcernToUse).deleteMany(removeQuery, options);
 				}
 
 				maybeEmitEvent(new AfterDeleteEvent<T>(queryObject, entityClass, collectionName));
@@ -3230,4 +3231,147 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 	public MongoDbFactory getMongoDbFactory() {
 		return mongoDbFactory;
 	}
+
+	/**
+	 * {@link BatchAggregationLoader} is a little helper that can process cursor results returned by an aggregation
+	 * command execution. On presence of a {@literal nextBatch} indicated by presence of an {@code id} field in the
+	 * {@code cursor} another {@code getMore} command gets executed reading the next batch of documents until all results
+	 * are loaded.
+	 *
+	 * @author Christoph Strobl
+	 * @since 1.10
+	 */
+	static class BatchAggregationLoader {
+
+		private static final String CURSOR_FIELD = "cursor";
+		private static final String RESULT_FIELD = "result";
+		private static final String BATCH_SIZE_FIELD = "batchSize";
+		private static final String FIRST_BATCH = "firstBatch";
+		private static final String NEXT_BATCH = "nextBatch";
+		private static final String SERVER_USED = "serverUsed";
+		private static final String OK = "ok";
+
+		private final MongoTemplate template;
+		private final ReadPreference readPreference;
+		private final int batchSize;
+
+		BatchAggregationLoader(MongoTemplate template, ReadPreference readPreference, int batchSize) {
+
+			this.template = template;
+			this.readPreference = readPreference;
+			this.batchSize = batchSize;
+		}
+
+		/**
+		 * Run aggregation command and fetch all results.
+		 */
+		Document aggregate(String collectionName, Aggregation aggregation, AggregationOperationContext context) {
+
+			Document command = prepareAggregationCommand(collectionName, aggregation, context, batchSize);
+
+			if (LOGGER.isDebugEnabled()) {
+				LOGGER.debug("Executing aggregation: {}", serializeToJsonSafely(command));
+			}
+
+			return mergeAggregationResults(aggregateBatched(command, collectionName, batchSize));
+		}
+
+		/**
+		 * Pre process the aggregation command sent to the server by adding {@code cursor} options to match execution on
+		 * different server versions.
+		 */
+		private static Document prepareAggregationCommand(String collectionName, Aggregation aggregation,
+				@Nullable AggregationOperationContext context, int batchSize) {
+
+			AggregationOperationContext rootContext = context == null ? Aggregation.DEFAULT_CONTEXT : context;
+			Document command = aggregation.toDocument(collectionName, rootContext);
+
+			if (!aggregation.getOptions().isExplain()) {
+				command.put(CURSOR_FIELD, new Document(BATCH_SIZE_FIELD, batchSize));
+			}
+
+			return command;
+		}
+
+		private List<Document> aggregateBatched(Document command, String collectionName, int batchSize) {
+
+			List<Document> results = new ArrayList<>();
+
+			Document commandResult = template.executeCommand(command, readPreference);
+			results.add(postProcessResult(commandResult));
+
+			while (hasNext(commandResult)) {
+
+				Document getMore = new Document("getMore", getNextBatchId(commandResult)) //
+						.append("collection", collectionName) //
+						.append(BATCH_SIZE_FIELD, batchSize);
+
+				commandResult = template.executeCommand(getMore, this.readPreference);
+				results.add(postProcessResult(commandResult));
+			}
+
+			return results;
+		}
+
+		private static Document postProcessResult(Document commandResult) {
+
+			if (!commandResult.containsKey(CURSOR_FIELD)) {
+				return commandResult;
+			}
+
+			Document resultObject = new Document(SERVER_USED, commandResult.get(SERVER_USED));
+			resultObject.put(OK, commandResult.get(OK));
+
+			Document cursor = (Document) commandResult.get(CURSOR_FIELD);
+			if (cursor.containsKey(FIRST_BATCH)) {
+				resultObject.put(RESULT_FIELD, cursor.get(FIRST_BATCH));
+			} else {
+				resultObject.put(RESULT_FIELD, cursor.get(NEXT_BATCH));
+			}
+
+			return resultObject;
+		}
+
+		private static Document mergeAggregationResults(List<Document> batchResults) {
+
+			if (batchResults.size() == 1) {
+				return batchResults.iterator().next();
+			}
+
+			Document commandResult = new Document();
+			List<Object> allResults = new ArrayList<>();
+
+			for (Document batchResult : batchResults) {
+
+				Collection documents = (Collection<?>) batchResult.get(RESULT_FIELD);
+				if (!CollectionUtils.isEmpty(documents)) {
+					allResults.addAll(documents);
+				}
+			}
+
+			// take general info from first batch
+			commandResult.put(SERVER_USED, batchResults.iterator().next().get(SERVER_USED));
+			commandResult.put(OK, batchResults.iterator().next().get(OK));
+
+			// and append the merged batchResults
+			commandResult.put(RESULT_FIELD, allResults);
+
+			return commandResult;
+		}
+
+		private static boolean hasNext(Document commandResult) {
+
+			if (!commandResult.containsKey(CURSOR_FIELD)) {
+				return false;
+			}
+
+			Object next = getNextBatchId(commandResult);
+			return next != null && ((Number) next).longValue() != 0L;
+		}
+
+		@Nullable
+		private static Object getNextBatchId(Document commandResult) {
+			return ((Document) commandResult.get(CURSOR_FIELD)).get("id");
+		}
+	}
 }

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java

@@ -114,6 +114,7 @@ import com.mongodb.MongoException;
 import com.mongodb.ReadPreference;
 import com.mongodb.WriteConcern;
 import com.mongodb.client.model.CreateCollectionOptions;
+import com.mongodb.client.model.DeleteOptions;
 import com.mongodb.client.model.Filters;
 import com.mongodb.client.model.FindOneAndDeleteOptions;
 import com.mongodb.client.model.FindOneAndUpdateOptions;
@@ -1680,27 +1681,34 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 
 		return execute(collectionName, collection -> {
 
-			maybeEmitEvent(new BeforeDeleteEvent<T>(queryObject, entityClass, collectionName));
+			Document removeQuey = queryMapper.getMappedObject(queryObject, entity);
 
-			Document dboq = queryMapper.getMappedObject(queryObject, entity);
+			maybeEmitEvent(new BeforeDeleteEvent<T>(removeQuey, entityClass, collectionName));
 
 			MongoAction mongoAction = new MongoAction(writeConcern, MongoActionOperation.REMOVE, collectionName, entityClass,
-					null, queryObject);
+					null, removeQuey);
+
+			final DeleteOptions deleteOptions = new DeleteOptions();
+			query.getCollation().map(Collation::toMongoCollation).ifPresent(deleteOptions::collation);
+
 			WriteConcern writeConcernToUse = prepareWriteConcern(mongoAction);
 			MongoCollection<Document> collectionToUse = prepareCollection(collection, writeConcernToUse);
 
 			if (LOGGER.isDebugEnabled()) {
 				LOGGER.debug("Remove using query: {} in collection: {}.",
-						new Object[] { serializeToJsonSafely(dboq), collectionName });
+						new Object[] { serializeToJsonSafely(removeQuey), collectionName });
 			}
 
-			query.getCollation().ifPresent(val -> {
+			if (query.getLimit() > 0 || query.getSkip() > 0) {
 
-				// TODO: add collation support as soon as it's there! See https://jira.mongodb.org/browse/JAVARS-27
-				throw new IllegalArgumentException("DeleteMany does currently not accept collation settings.");
-			});
-
-			return collectionToUse.deleteMany(dboq);
+				FindPublisher<Document> cursor = new QueryFindPublisherPreparer(query, entityClass)
+						.prepare(collection.find(removeQuey)).projection(new Document(ID_FIELD, 1));
+				return Flux.from(cursor).map(doc -> doc.get(ID_FIELD)).collectList().flatMap(val -> {
+					return Mono.from(collectionToUse.deleteMany(new Document(ID_FIELD, new Document("$in", val)), deleteOptions));
+				});
+			} else {
+				return collectionToUse.deleteMany(removeQuey, deleteOptions);
+			}
 
 		}).doOnNext(deleteResult -> maybeEmitEvent(new AfterDeleteEvent<T>(queryObject, entityClass, collectionName)))
 				.next();

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateTests.java

@@ -99,6 +99,7 @@ import com.mongodb.client.FindIterable;
 import com.mongodb.client.ListIndexesIterable;
 import com.mongodb.client.MongoCollection;
 import com.mongodb.client.MongoCursor;
+import com.mongodb.client.result.DeleteResult;
 import com.mongodb.client.result.UpdateResult;
 
 /**
@@ -3326,6 +3327,34 @@ public class MongoTemplateTests {
 		assertThat(template.find(new Query().limit(1), Sample.class)).hasSize(1);
 	}
 
+	@Test // DATAMONGO-1870
+	public void removeShouldConsiderLimit() {
+
+		for (int i = 0; i < 100; i++) {
+			template.save(new Sample("id-" + i, i % 2 == 0 ? "stark" : "lannister"));
+		}
+
+		DeleteResult wr = template.remove(query(where("field").is("lannister")).limit(25), Sample.class);
+
+		assertThat(wr.getDeletedCount()).isEqualTo(25L);
+		assertThat(template.count(new Query(), Sample.class)).isEqualTo(75L);
+	}
+
+	@Test // DATAMONGO-1870
+	public void removeShouldConsiderSkipAndSort() {
+
+		for (int i = 0; i < 100; i++) {
+			template.save(new Sample("id-" + i, i % 2 == 0 ? "stark" : "lannister"));
+		}
+
+		DeleteResult wr = template.remove(new Query().skip(25).with(Sort.by("field")), Sample.class);
+
+		assertThat(wr.getDeletedCount()).isEqualTo(75L);
+		assertThat(template.count(new Query(), Sample.class)).isEqualTo(25L);
+		assertThat(template.count(query(where("field").is("lannister")), Sample.class)).isEqualTo(25L);
+		assertThat(template.count(query(where("field").is("stark")), Sample.class)).isEqualTo(0L);
+	}
+
 	static class TypeWithNumbers {
 
 		@Id String id;

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/MongoTemplateUnitTests.java

@@ -164,7 +164,7 @@ public class MongoTemplateUnitTests extends MongoOperationsUnitTests {
 		new MongoTemplate(null, "database");
 	}
 
-	@Test(expected = DataAccessException.class)
+	@Test(expected = IllegalArgumentException.class) // DATAMONGO-1870
 	public void removeHandlesMongoExceptionProperly() throws Exception {
 
 		MongoTemplate template = mockOutGetDb();

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateTests.java

@@ -37,6 +37,7 @@ import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
+import org.assertj.core.api.Assertions;
 import org.assertj.core.api.Assumptions;
 import org.bson.BsonDocument;
 import org.bson.Document;
@@ -66,7 +67,6 @@ import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.Update;
-import org.springframework.data.mongodb.test.util.Assertions;
 import org.springframework.data.mongodb.test.util.ReplicaSet;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -1124,6 +1124,35 @@ public class ReactiveMongoTemplateTests {
 
 	}
 
+	@Test // DATAMONGO-1870
+	public void removeShouldConsiderLimit() {
+
+		for (int i = 0; i < 100; i++) {
+			StepVerifier.create(template.save(new Sample("id-" + i, i % 2 == 0 ? "stark" : "lannister"))).expectNextCount(1)
+					.verifyComplete();
+		}
+
+		StepVerifier.create(template.remove(query(where("field").is("lannister")).limit(25), Sample.class))
+				.assertNext(wr -> Assertions.assertThat(wr.getDeletedCount()).isEqualTo(25L)).verifyComplete();
+	}
+
+	@Test // DATAMONGO-1870
+	public void removeShouldConsiderSkipAndSort() {
+
+		for (int i = 0; i < 100; i++) {
+			StepVerifier.create(template.save(new Sample("id-" + i, i % 2 == 0 ? "stark" : "lannister"))).expectNextCount(1)
+					.verifyComplete();
+		}
+
+		StepVerifier.create(template.remove(new Query().skip(25).with(Sort.by("field")), Sample.class))
+				.assertNext(wr -> Assertions.assertThat(wr.getDeletedCount()).isEqualTo(75L)).verifyComplete();
+
+		StepVerifier.create(template.count(query(where("field").is("lannister")), Sample.class)).expectNext(25L)
+				.verifyComplete();
+		StepVerifier.create(template.count(query(where("field").is("stark")), Sample.class)).expectNext(0L)
+				.verifyComplete();
+	}
+
 	private PersonWithAList createPersonWithAList(String firstname, int age) {
 
 		PersonWithAList p = new PersonWithAList();
@@ -1146,5 +1175,4 @@ public class ReactiveMongoTemplateTests {
 			this.field = field;
 		}
 	}
-
 }

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/ReactiveMongoTemplateUnitTests.java

@@ -169,16 +169,16 @@ public class ReactiveMongoTemplateUnitTests {
 		assertThat(options.getValue().getCollation().getLocale(), is("fr"));
 	}
 
-	@Ignore("see https://jira.mongodb.org/browse/JAVARS-27")
 	@Test // DATAMONGO-1518
 	public void findAndRemoveManyShouldUseCollationWhenPresent() {
 
+		when(collection.deleteMany(any(Bson.class), any())).thenReturn(Mono.empty());
+
 		template.doRemove("collection-1", new BasicQuery("{}").collation(Collation.of("fr")), AutogenerateableId.class)
 				.subscribe();
 
 		ArgumentCaptor<DeleteOptions> options = ArgumentCaptor.forClass(DeleteOptions.class);
-		// the current mongodb-driver-reactivestreams:1.4.0 driver does not offer deleteMany with options.
-		// verify(collection).deleteMany(Mockito.any(), options.capture());
+		verify(collection).deleteMany(Mockito.any(), options.capture());
 
 		assertThat(options.getValue().getCollation().getLocale(), is("fr"));
 	}

src/main/asciidoc/reference/mongodb.adoc

@@ -951,7 +951,25 @@ assertThat(p.getAge(), is(1));
 
 You can use several overloaded methods to remove an object from the database.
 
-* *remove* Remove the given document based on one of the following: a specific object instance, a query document criteria combined with a class or a query document criteria combined with a specific collection name.
+====
+[source,java]
+----
+template.remove(tywin, "GOT");                                              <1>
+
+template.remove(query(where("lastname").is("lannister")), "GOT");           <2>
+
+template.remove(new Query().limit(3), "GOT");                               <3>
+
+template.findAllAndRemove(query(where("lastname").is("lannister"), "GOT");  <4>
+
+template.findAllAndRemove(new Query().limit(3), "GOT");                     <5>
+----
+<1> Remove a single entity via its `id` from the associated collection.
+<2> Remove all documents matching the criteria of the query from the `GOT` collection.
+<3> Rewmove the first 3 documents in the `GOT` collection. Unlike <2> the documents to remove are identified via their `id` using the given query applying `sort`, `limit` and `skip` options and then removed all at once in a seperate step.
+<4> Remove all documents matching the criteria of the query from the `GOT` collection. Unlike <3> documents do not get deleted in a batch but one by one.
+<5> Remove the first 3 documents in the `GOT` collection. Unlike <3> documents do not get deleted in a batch but one by one.
+====
 
 [[mongo-template.optimistic-locking]]
 === Optimistic locking