Add support to create views via reactive/template API.

This commit introduces support to create MongoDB Views directly via the Reactive-/MongoOperations API.

Closes: #2594
Original pull request: #4142.

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

@@ -26,7 +26,9 @@ import org.bson.Document;
 import org.springframework.data.geo.GeoResults;
 import org.springframework.data.mongodb.core.BulkOperations.BulkMode;
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
+import org.springframework.data.mongodb.core.aggregation.AggregationOperation;
 import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
+import org.springframework.data.mongodb.core.aggregation.AggregationPipeline;
 import org.springframework.data.mongodb.core.aggregation.AggregationResults;
 import org.springframework.data.mongodb.core.aggregation.AggregationUpdate;
 import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
@@ -280,6 +282,56 @@ public interface MongoOperations extends FluentMongoOperations {
 	 */
 	MongoCollection<Document> createCollection(String collectionName, @Nullable CollectionOptions collectionOptions);
 
+	/**
+	 * Create a view with the the provided name whose contents are defined by the {@link AggregationOperation pipeline
+	 * stages} on another collection or view identified by the given {@link #getCollectionName(Class) source type}.
+	 * 
+	 * @param name the name of the view to create.
+	 * @param source the type defining the views source collection.
+	 * @param stages the {@link AggregationOperation aggregation pipeline stages} defining the view content.
+	 * @since 4.0
+	 */
+	default MongoCollection<Document> createView(String name, Class<?> source, AggregationOperation... stages) {
+		return createView(name, source, AggregationPipeline.of(stages));
+	}
+
+	/**
+	 * Create a view with the the provided name whose contents are defined by the  {@link AggregationPipeline pipeline} on
+	 * another collection or view identified by the given {@link #getCollectionName(Class) source type}.
+	 *
+	 * @param name the name of the view to create.
+	 * @param source the type defining the views source collection.
+	 * @param pipeline the {@link AggregationPipeline} defining the view content.
+	 * @since 4.0
+	 */
+	default MongoCollection<Document> createView(String name, Class<?> source, AggregationPipeline pipeline) {
+		return createView(name, source, pipeline, null);
+	}
+
+	/**
+	 * Create a view with the the provided name whose contents are defined by the {@link AggregationPipeline pipeline} on
+	 * another collection or view identified by the given {@link #getCollectionName(Class) source type}.
+	 *
+	 * @param name the name of the view to create.
+	 * @param source the type defining the views source collection.
+	 * @param pipeline the {@link AggregationPipeline} defining the view content.
+	 * @param options additional settings to apply when creating the view. Can be {@literal null}.
+	 * @since 4.0
+	 */
+	MongoCollection<Document> createView(String name, Class<?> source, AggregationPipeline pipeline, @Nullable ViewOptions options);
+
+	/**
+	 * Create a view with the the provided name whose contents are defined by the {@link AggregationPipeline pipeline} on
+	 * another collection or view identified by the given source.
+	 *
+	 * @param name the name of the view to create.
+	 * @param source the name of the collection or view defining the to be created views source.
+	 * @param pipeline the {@link AggregationPipeline} defining the view content.
+	 * @param options additional settings to apply when creating the view. Can be {@literal null}.
+	 * @since 4.0
+	 */
+	MongoCollection<Document> createView(String name, String source, AggregationPipeline pipeline, @Nullable ViewOptions options);
+
 	/**
 	 * A set of collection names.
 	 *

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

@@ -67,6 +67,7 @@ import org.springframework.data.mongodb.core.QueryOperations.UpdateContext;
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
 import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
 import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
+import org.springframework.data.mongodb.core.aggregation.AggregationPipeline;
 import org.springframework.data.mongodb.core.aggregation.AggregationResults;
 import org.springframework.data.mongodb.core.aggregation.TypedAggregation;
 import org.springframework.data.mongodb.core.convert.DbRefResolver;
@@ -632,6 +633,40 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 				operations.convertToCreateCollectionOptions(collectionOptions, Object.class));
 	}
 
+	@Override
+	public MongoCollection<Document> createView(String name, Class<?> source, AggregationPipeline pipeline, @Nullable ViewOptions options) {
+
+		return createView(name, getCollectionName(source),
+				queryOperations.createAggregation(Aggregation.newAggregation(source, pipeline.getOperations()), source),
+				options);
+	}
+
+	@Override
+	public MongoCollection<Document> createView(String name, String source, AggregationPipeline pipeline, @Nullable ViewOptions options) {
+
+		return createView(name, source,
+				queryOperations.createAggregation(Aggregation.newAggregation(pipeline.getOperations()), (Class<?>) null),
+				options);
+	}
+
+	private MongoCollection<Document> createView(String name, String source, AggregationDefinition aggregation,
+			@Nullable ViewOptions options) {
+		return doCreateView(name, source, aggregation.getAggregationPipeline(), options);
+	}
+
+	protected MongoCollection<Document> doCreateView(String name, String source, List<Document> pipeline, @Nullable ViewOptions options) {
+
+		CreateViewOptions viewOptions = new CreateViewOptions();
+		if (options != null) {
+			options.getCollation().map(Collation::toMongoCollation).ifPresent(viewOptions::collation);
+		}
+
+		return execute(db -> {
+			db.createView(name, source, pipeline, viewOptions);
+			return db.getCollection(name);
+		});
+	}
+
 	@Override
 	@SuppressWarnings("ConstantConditions")
 	public MongoCollection<Document> getCollection(String collectionName) {

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

@@ -15,6 +15,8 @@
  */
 package org.springframework.data.mongodb.core;
 
+import org.springframework.data.mongodb.core.aggregation.AggregationOperation;
+import org.springframework.data.mongodb.core.aggregation.AggregationPipeline;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -240,6 +242,56 @@ public interface ReactiveMongoOperations extends ReactiveFluentMongoOperations {
 	 */
 	Mono<MongoCollection<Document>> createCollection(String collectionName, CollectionOptions collectionOptions);
 
+	/**
+	 * Create a view with the the provided name whose contents are defined by the {@link AggregationOperation pipeline
+	 * stages} on another collection or view identified by the given {@link #getCollectionName(Class) source type}.
+	 *
+	 * @param name the name of the view to create.
+	 * @param source the type defining the views source collection.
+	 * @param stages the {@link AggregationOperation aggregation pipeline stages} defining the view content.
+	 * @since 4.0
+	 */
+	default Mono<MongoCollection<Document>> createView(String name, Class<?> source, AggregationOperation... stages) {
+		return createView(name, source, AggregationPipeline.of(stages));
+	}
+
+	/**
+	 * Create a view with the the provided name whose contents are defined by the  {@link AggregationPipeline pipeline} on
+	 * another collection or view identified by the given {@link #getCollectionName(Class) source type}.
+	 *
+	 * @param name the name of the view to create.
+	 * @param source the type defining the views source collection.
+	 * @param pipeline the {@link AggregationPipeline} defining the view content.
+	 * @since 4.0
+	 */
+	default Mono<MongoCollection<Document>> createView(String name, Class<?> source, AggregationPipeline pipeline) {
+		return createView(name, source, pipeline, null);
+	}
+
+	/**
+	 * Create a view with the the provided name whose contents are defined by the {@link AggregationPipeline pipeline} on
+	 * another collection or view identified by the given {@link #getCollectionName(Class) source type}.
+	 *
+	 * @param name the name of the view to create.
+	 * @param source the type defining the views source collection.
+	 * @param pipeline the {@link AggregationPipeline} defining the view content.
+	 * @param options additional settings to apply when creating the view. Can be {@literal null}.
+	 * @since 4.0
+	 */
+	Mono<MongoCollection<Document>> createView(String name, Class<?> source, AggregationPipeline pipeline, @Nullable ViewOptions options);
+
+	/**
+	 * Create a view with the the provided name whose contents are defined by the {@link AggregationPipeline pipeline} on
+	 * another collection or view identified by the given source.
+	 *
+	 * @param name the name of the view to create.
+	 * @param source the name of the collection or view defining the to be created views source.
+	 * @param pipeline the {@link AggregationPipeline} defining the view content.
+	 * @param options additional settings to apply when creating the view. Can be {@literal null}.
+	 * @since 4.0
+	 */
+	Mono<MongoCollection<Document>> createView(String name, String source, AggregationPipeline pipeline, @Nullable ViewOptions options);
+
 	/**
 	 * A set of collection names.
 	 *

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

@@ -46,7 +46,6 @@ import org.bson.conversions.Bson;
 import org.bson.types.ObjectId;
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
-
 import org.springframework.beans.BeansException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
@@ -81,6 +80,7 @@ import org.springframework.data.mongodb.core.QueryOperations.UpdateContext;
 import org.springframework.data.mongodb.core.aggregation.Aggregation;
 import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
 import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
+import org.springframework.data.mongodb.core.aggregation.AggregationPipeline;
 import org.springframework.data.mongodb.core.aggregation.PrefixingDelegatingAggregationOperationContext;
 import org.springframework.data.mongodb.core.aggregation.RelaxedTypeBasedAggregationOperationContext;
 import org.springframework.data.mongodb.core.aggregation.TypeBasedAggregationOperationContext;
@@ -127,16 +127,7 @@ import com.mongodb.CursorType;
 import com.mongodb.MongoException;
 import com.mongodb.ReadPreference;
 import com.mongodb.WriteConcern;
-import com.mongodb.client.model.CountOptions;
-import com.mongodb.client.model.CreateCollectionOptions;
-import com.mongodb.client.model.DeleteOptions;
-import com.mongodb.client.model.EstimatedDocumentCountOptions;
-import com.mongodb.client.model.FindOneAndDeleteOptions;
-import com.mongodb.client.model.FindOneAndReplaceOptions;
-import com.mongodb.client.model.FindOneAndUpdateOptions;
-import com.mongodb.client.model.ReplaceOptions;
-import com.mongodb.client.model.ReturnDocument;
-import com.mongodb.client.model.UpdateOptions;
+import com.mongodb.client.model.*;
 import com.mongodb.client.model.changestream.FullDocument;
 import com.mongodb.client.result.DeleteResult;
 import com.mongodb.client.result.InsertOneResult;
@@ -671,6 +662,42 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 		return doCreateCollection(collectionName, convertToCreateCollectionOptions(collectionOptions));
 	}
 
+	@Override
+	public Mono<MongoCollection<Document>> createView(String name, Class<?> source, AggregationPipeline pipeline, @Nullable ViewOptions options) {
+
+		return createView(name, getCollectionName(source),
+				queryOperations.createAggregation(Aggregation.newAggregation(source, pipeline.getOperations()), source),
+				options);
+	}
+
+	@Override
+	public Mono<MongoCollection<Document>> createView(String name, String source, AggregationPipeline pipeline,
+			@Nullable ViewOptions options) {
+
+		return createView(name, source,
+				queryOperations.createAggregation(Aggregation.newAggregation(pipeline.getOperations()), (Class<?>) null),
+				options);
+	}
+
+	private Mono<MongoCollection<Document>> createView(String name, String source, AggregationDefinition aggregation,
+			@Nullable ViewOptions options) {
+		return doCreateView(name, source, aggregation.getAggregationPipeline(), options);
+	}
+
+	protected Mono<MongoCollection<Document>> doCreateView(String name, String source, List<Document> pipeline,
+			@Nullable ViewOptions options) {
+
+		CreateViewOptions viewOptions = new CreateViewOptions();
+		if (options != null) {
+			options.getCollation().map(Collation::toMongoCollation).ifPresent(viewOptions::collation);
+		}
+
+		return execute(db -> {
+			return Flux.from(db.createView(name, source, pipeline, viewOptions))
+					.then(Mono.fromSupplier(() -> db.getCollection(name)));
+		}).next();
+	}
+
 	@Override
 	public Mono<MongoCollection<Document>> getCollection(String collectionName) {
 

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

@@ -0,0 +1,65 @@
+/*
+ * 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.mongodb.core;
+
+import java.util.Optional;
+
+import org.springframework.data.mongodb.core.query.Collation;
+import org.springframework.lang.Nullable;
+
+/**
+ * Immutable object holding additional options to be applied when creating a MongoDB
+ * <a href="https://www.mongodb.com/docs/manual/core/views/">views</a>.
+ * 
+ * @author Christoph Strobl
+ * @since 4.0
+ */
+public class ViewOptions {
+
+	@Nullable private Collation collation;
+
+	static ViewOptions none() {
+		return new ViewOptions();
+	}
+
+	/**
+	 * Creates new instance of {@link ViewOptions}.
+	 */
+	public ViewOptions() {
+		this(null);
+	}
+
+	private ViewOptions(@Nullable Collation collation) {
+		this.collation = collation;
+	}
+
+	/**
+	 * Get the {@link Collation} to be set.
+	 *
+	 * @return {@link Optional#empty()} if not set.
+	 */
+	public Optional<Collation> getCollation() {
+		return Optional.ofNullable(collation);
+	}
+
+	/**
+	 * @param collation the {@link Collation} to use for language-specific string comparison.
+	 * @return new instance of {@link ViewOptions}.
+	 */
+	public ViewOptions collation(Collation collation) {
+		return new ViewOptions(collation);
+	}
+}

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

@@ -16,6 +16,7 @@
 package org.springframework.data.mongodb.core.aggregation;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.function.Predicate;
@@ -34,6 +35,10 @@ public class AggregationPipeline {
 
 	private final List<AggregationOperation> pipeline;
 
+	public static AggregationPipeline of(AggregationOperation... stages) {
+		return new AggregationPipeline(Arrays.asList(stages));
+	}
+
 	/**
 	 * Create an empty pipeline
 	 */

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

@@ -0,0 +1,158 @@
+/*
+ * 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.mongodb.core;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.springframework.data.mongodb.core.aggregation.Aggregation.*;
+import static org.springframework.data.mongodb.core.query.Criteria.*;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.bson.Document;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.data.mongodb.core.aggregation.AggregationPipeline;
+import org.springframework.data.mongodb.core.mapping.Field;
+import org.springframework.data.mongodb.core.query.Collation;
+import org.springframework.data.mongodb.test.util.Client;
+import org.springframework.data.mongodb.test.util.CollectionInfo;
+import org.springframework.data.mongodb.test.util.MongoClientExtension;
+import org.springframework.data.mongodb.test.util.MongoTestUtils;
+
+import com.mongodb.client.MongoClient;
+
+/**
+ * @author Christoph Strobl
+ */
+@ExtendWith(MongoClientExtension.class)
+public class MongoTemplateViewTests {
+
+	static @Client MongoClient client;
+	static final String DB_NAME = "mongo-template-view-tests";
+
+	private MongoTemplate template;
+
+	Student alex = new Student(22001L, "Alex", 1, 4.0D);
+	Student bernie = new Student(21001L, "bernie", 2, 3.7D);
+	Student chris = new Student(20010L, "Chris", 3, 2.5D);
+	Student drew = new Student(22021L, "Drew", 1, 3.2D);
+	Student harley1 = new Student(17301L, "harley", 6, 3.1D);
+	Student farmer = new Student(21022L, "Farmer", 1, 2.2D);
+	Student george = new Student(20020L, "george", 3, 2.8D);
+	Student harley2 = new Student(18020, "Harley", 5, 2.8D);
+
+	List<Student> students = Arrays.asList(alex, bernie, chris, drew, harley1, farmer, george, harley2);
+
+	@BeforeEach
+	void beforeEach() {
+		template = new MongoTemplate(client, DB_NAME);
+	}
+
+	@AfterEach
+	void afterEach() {
+		client.getDatabase(DB_NAME).drop();
+	}
+
+	@Test // GH-2594
+	void createsViewFromPipeline() {
+
+		template.insertAll(students);
+
+		template.createView("firstYears", Student.class, match(where("year").is(1)));
+
+		CollectionInfo collectionInfo = MongoTestUtils.readCollectionInfo(client.getDatabase(DB_NAME), "firstYears");
+		assertThat(collectionInfo.isView()).isTrue();
+		assertThat(collectionInfo.getViewTarget()).isEqualTo("student");
+		assertThat(collectionInfo.getViewPipeline()).containsExactly(new Document("$match", new Document("year", 1)));
+	}
+
+	@Test // GH-2594
+	void mapsPipelineAgainstDomainObject() {
+
+		template.insertAll(students);
+
+		template.createView("fakeStudents", Student.class, match(where("studentID").gte("22")));
+
+		CollectionInfo collectionInfo = MongoTestUtils.readCollectionInfo(client.getDatabase(DB_NAME), "fakeStudents");
+		assertThat(collectionInfo.isView()).isTrue();
+		assertThat(collectionInfo.getViewPipeline())
+				.containsExactly(new Document("$match", new Document("sID", new Document("$gte", "22"))));
+	}
+
+	@Test // GH-2594
+	void takesPipelineAsIsIfNoTypeDefined() {
+
+		template.insertAll(students);
+
+		template.createView("fakeStudents", "student", AggregationPipeline.of(match(where("studentID").gte("22"))),
+				ViewOptions.none());
+
+		CollectionInfo collectionInfo = MongoTestUtils.readCollectionInfo(client.getDatabase(DB_NAME), "fakeStudents");
+		assertThat(collectionInfo.isView()).isTrue();
+		assertThat(collectionInfo.getViewPipeline())
+				.containsExactly(new Document("$match", new Document("studentID", new Document("$gte", "22"))));
+	}
+
+	@Test // GH-2594
+	void readsFromView() {
+
+		template.insertAll(students);
+		client.getDatabase(DB_NAME).createView("firstYears", "student",
+				Arrays.asList(new Document("$match", new Document("year", 1))));
+
+		assertThat(template.query(Student.class).inCollection("firstYears").all()).containsExactlyInAnyOrder(alex, drew,
+				farmer);
+	}
+
+	@Test // GH-2594
+	void appliesCollationToView() {
+
+		template.insertAll(students);
+
+		template.createView("firstYears", Student.class, AggregationPipeline.of(match(where("year").is(1))),
+				new ViewOptions().collation(Collation.of("en_US")));
+
+		CollectionInfo collectionInfo = MongoTestUtils.readCollectionInfo(client.getDatabase(DB_NAME), "firstYears");
+		assertThat(collectionInfo.isView()).isTrue();
+		assertThat(collectionInfo.getCollation().getLocale()).isEqualTo("en_US");
+	}
+
+	@Data
+	@NoArgsConstructor
+	private static class Student {
+
+		@Field("sID") Long studentID;
+
+		int year;
+
+		double score;
+
+		String name;
+
+		public Student(long studentID, String name, int year, double score) {
+			this.studentID = studentID;
+			this.name = name;
+			this.year = year;
+			this.score = score;
+		}
+	}
+}

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

@@ -0,0 +1,162 @@
+/*
+ * 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.mongodb.core;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.springframework.data.mongodb.core.aggregation.Aggregation.*;
+import static org.springframework.data.mongodb.core.query.Criteria.*;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import reactor.test.StepVerifier;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.bson.Document;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.data.mongodb.core.aggregation.AggregationPipeline;
+import org.springframework.data.mongodb.core.mapping.Field;
+import org.springframework.data.mongodb.core.query.Collation;
+import org.springframework.data.mongodb.test.util.Client;
+import org.springframework.data.mongodb.test.util.CollectionInfo;
+import org.springframework.data.mongodb.test.util.MongoClientExtension;
+import org.springframework.data.mongodb.test.util.MongoTestUtils;
+
+import com.mongodb.reactivestreams.client.MongoClient;
+
+/**
+ * @author Christoph Strobl
+ */
+@ExtendWith(MongoClientExtension.class)
+public class ReactiveMongoTemplateViewTests {
+
+	static @Client com.mongodb.client.MongoClient client;
+	static @Client MongoClient reactiveClient;
+	static final String DB_NAME = "reactive-mongo-template-view-tests";
+
+	private ReactiveMongoTemplate template;
+
+	Student alex = new Student(22001L, "Alex", 1, 4.0D);
+	Student bernie = new Student(21001L, "bernie", 2, 3.7D);
+	Student chris = new Student(20010L, "Chris", 3, 2.5D);
+	Student drew = new Student(22021L, "Drew", 1, 3.2D);
+	Student harley1 = new Student(17301L, "harley", 6, 3.1D);
+	Student farmer = new Student(21022L, "Farmer", 1, 2.2D);
+	Student george = new Student(20020L, "george", 3, 2.8D);
+	Student harley2 = new Student(18020, "Harley", 5, 2.8D);
+
+	List<Student> students = Arrays.asList(alex, bernie, chris, drew, harley1, farmer, george, harley2);
+
+	@BeforeEach
+	void beforeEach() {
+		template = new ReactiveMongoTemplate(reactiveClient, DB_NAME);
+	}
+
+	@AfterEach
+	void afterEach() {
+		client.getDatabase(DB_NAME).drop();
+	}
+
+	@Test // GH-2594
+	void createsViewFromPipeline() {
+
+		template.insertAll(students).then().as(StepVerifier::create).verifyComplete();
+
+		template.createView("firstYears", Student.class, match(where("year").is(1))).then().as(StepVerifier::create)
+				.verifyComplete();
+
+		CollectionInfo collectionInfo = MongoTestUtils.readCollectionInfo(client.getDatabase(DB_NAME), "firstYears");
+		assertThat(collectionInfo.isView()).isTrue();
+		assertThat(collectionInfo.getViewTarget()).isEqualTo("student");
+		assertThat(collectionInfo.getViewPipeline()).containsExactly(new Document("$match", new Document("year", 1)));
+	}
+
+	@Test // GH-2594
+	void mapsPipelineAgainstDomainObject() {
+
+		template.insertAll(students).then().as(StepVerifier::create).verifyComplete();
+
+		template.createView("fakeStudents", Student.class, match(where("studentID").gte("22"))).then()
+				.as(StepVerifier::create).verifyComplete();
+
+		CollectionInfo collectionInfo = MongoTestUtils.readCollectionInfo(client.getDatabase(DB_NAME), "fakeStudents");
+		assertThat(collectionInfo.isView()).isTrue();
+		assertThat(collectionInfo.getViewPipeline())
+				.containsExactly(new Document("$match", new Document("sID", new Document("$gte", "22"))));
+	}
+
+	@Test // GH-2594
+	void takesPipelineAsIsIfNoTypeDefined() {
+
+		template.insertAll(students).then().as(StepVerifier::create).verifyComplete();
+
+		template.createView("fakeStudents", "student", AggregationPipeline.of(match(where("studentID").gte("22"))),
+				ViewOptions.none()).then().as(StepVerifier::create).verifyComplete();
+
+		CollectionInfo collectionInfo = MongoTestUtils.readCollectionInfo(client.getDatabase(DB_NAME), "fakeStudents");
+		assertThat(collectionInfo.isView()).isTrue();
+		assertThat(collectionInfo.getViewPipeline())
+				.containsExactly(new Document("$match", new Document("studentID", new Document("$gte", "22"))));
+	}
+
+	@Test // GH-2594
+	void readsFromView() {
+
+		template.insertAll(students).then().as(StepVerifier::create).verifyComplete();
+		client.getDatabase(DB_NAME).createView("firstYears", "student",
+				Arrays.asList(new Document("$match", new Document("year", 1))));
+
+		template.query(Student.class).inCollection("firstYears").all().collectList().as(StepVerifier::create)
+				.consumeNextWith(it -> assertThat(it).containsExactlyInAnyOrder(alex, drew, farmer));
+	}
+
+	@Test // GH-2594
+	void appliesCollationToView() {
+
+		template.insertAll(students).then().as(StepVerifier::create).verifyComplete();
+
+		template.createView("firstYears", Student.class, AggregationPipeline.of(match(where("year").is(1))),
+				new ViewOptions().collation(Collation.of("en_US"))).then().as(StepVerifier::create).verifyComplete();
+
+		CollectionInfo collectionInfo = MongoTestUtils.readCollectionInfo(client.getDatabase(DB_NAME), "firstYears");
+		assertThat(collectionInfo.isView()).isTrue();
+		assertThat(collectionInfo.getCollation().getLocale()).isEqualTo("en_US");
+	}
+
+	@Data
+	@NoArgsConstructor
+	private static class Student {
+
+		@Field("sID") Long studentID;
+
+		int year;
+
+		double score;
+
+		String name;
+
+		public Student(long studentID, String name, int year, double score) {
+			this.studentID = studentID;
+			this.name = name;
+			this.year = year;
+			this.score = score;
+		}
+	}
+}

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/test/util/CollectionInfo.java

@@ -0,0 +1,98 @@
+/*
+ * 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.mongodb.test.util;
+
+import java.util.List;
+
+import org.bson.Document;
+import org.springframework.util.ObjectUtils;
+
+import com.mongodb.client.model.Collation;
+
+/**
+ * Value Object providing a methods for accessing collection/view information within a raw {@link Document}.
+ * 
+ * @author Christoph Strobl
+ */
+public class CollectionInfo {
+
+	private final Document source;
+
+	public static CollectionInfo from(Document source) {
+		return new CollectionInfo(source);
+	}
+
+	CollectionInfo(Document source) {
+		this.source = source;
+	}
+
+	/**
+	 * @return the collection/view name.
+	 */
+	public String getName() {
+		return source.getString("name");
+	}
+
+	/**
+	 * @return {@literal true} if the {@literal type} equals {@literal view}.
+	 */
+	public boolean isView() {
+		return ObjectUtils.nullSafeEquals("view", source.get("type"));
+	}
+
+	/**
+	 * @return the {@literal options.viewOn} value.
+	 * @throws IllegalStateException if not {@link #isView() a view}.
+	 */
+	public String getViewTarget() {
+
+		if (isView()) {
+			return getOptionValue("viewOn", String.class);
+		}
+		throw new IllegalStateException(getName() + " is not a view");
+	}
+
+	/**
+	 * @return the {@literal options.pipeline} value.
+	 * @throws IllegalStateException if not {@link #isView() a view}.
+	 */
+	public List<Document> getViewPipeline() {
+
+		if (isView()) {
+			return getOptions().getList("pipeline", Document.class);
+		}
+
+		throw new IllegalStateException(getName() + " is not a view");
+	}
+
+	/**
+	 * @return the {@literal options.collation} value.
+	 * @throws IllegalStateException if not {@link #isView() a view}.
+	 */
+	public Collation getCollation() {
+
+		return org.springframework.data.mongodb.core.query.Collation.from(getOptionValue("collation", Document.class))
+				.toMongoCollation();
+	}
+
+	private Document getOptions() {
+		return source.get("options", Document.class);
+	}
+
+	private <T> T getOptionValue(String key, Class<T> type) {
+		return getOptions().get(key, type);
+	}
+}

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/test/util/MongoTestUtils.java

@@ -17,6 +17,7 @@ package org.springframework.data.mongodb.test.util;
 
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
+import reactor.util.retry.Retry;
 
 import java.time.Duration;
 import java.util.List;
@@ -35,7 +36,6 @@ import com.mongodb.client.MongoClient;
 import com.mongodb.client.MongoCollection;
 import com.mongodb.client.MongoDatabase;
 import com.mongodb.reactivestreams.client.MongoClients;
-import reactor.util.retry.Retry;
 
 /**
  * Utility to create (and reuse) imperative and reactive {@code MongoClient} instances.
@@ -276,4 +276,15 @@ public class MongoTestUtils {
 			e.printStackTrace();
 		}
 	}
+
+	public static CollectionInfo readCollectionInfo(MongoDatabase db, String collectionName) {
+
+		List<Document> list = db.runCommand(new Document().append("listCollections", 1).append("filter", new Document("name", collectionName)))
+				.get("cursor", Document.class).get("firstBatch", List.class);
+
+		if(list.isEmpty()) {
+			throw new IllegalStateException(String.format("Collection %s not found.", collectionName));
+		}
+		return CollectionInfo.from(list.get(0));
+	}
 }