Add support for creating Time Series collection.

Introduce time series to CollectionOptions and add dedicated TimeSeries annotation to derive values from.

Closes #3731
Original pull request: #3732.

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

@@ -17,8 +17,11 @@ package org.springframework.data.mongodb.core;
 
 import java.util.Optional;
 
+import org.springframework.data.mongodb.core.mapping.Field;
 import org.springframework.data.mongodb.core.query.Collation;
 import org.springframework.data.mongodb.core.schema.MongoJsonSchema;
+import org.springframework.data.mongodb.core.timeseries.Granularities;
+import org.springframework.data.mongodb.core.timeseries.Granularity;
 import org.springframework.data.mongodb.core.validation.Validator;
 import org.springframework.data.util.Optionals;
 import org.springframework.lang.Nullable;
@@ -42,6 +45,7 @@ public class CollectionOptions {
 	private @Nullable Boolean capped;
 	private @Nullable Collation collation;
 	private ValidationOptions validationOptions;
+	private @Nullable TimeSeriesOptions timeSeriesOptions;
 
 	/**
 	 * Constructs a new <code>CollectionOptions</code> instance.
@@ -54,17 +58,19 @@ public class CollectionOptions {
 	 */
 	@Deprecated
 	public CollectionOptions(@Nullable Long size, @Nullable Long maxDocuments, @Nullable Boolean capped) {
-		this(size, maxDocuments, capped, null, ValidationOptions.none());
+		this(size, maxDocuments, capped, null, ValidationOptions.none(), null);
 	}
 
 	private CollectionOptions(@Nullable Long size, @Nullable Long maxDocuments, @Nullable Boolean capped,
-			@Nullable Collation collation, ValidationOptions validationOptions) {
+			@Nullable Collation collation, ValidationOptions validationOptions,
+			@Nullable TimeSeriesOptions timeSeriesOptions) {
 
 		this.maxDocuments = maxDocuments;
 		this.size = size;
 		this.capped = capped;
 		this.collation = collation;
 		this.validationOptions = validationOptions;
+		this.timeSeriesOptions = timeSeriesOptions;
 	}
 
 	/**
@@ -78,7 +84,7 @@ public class CollectionOptions {
 
 		Assert.notNull(collation, "Collation must not be null!");
 
-		return new CollectionOptions(null, null, null, collation, ValidationOptions.none());
+		return new CollectionOptions(null, null, null, collation, ValidationOptions.none(), null);
 	}
 
 	/**
@@ -88,7 +94,21 @@ public class CollectionOptions {
 	 * @since 2.0
 	 */
 	public static CollectionOptions empty() {
-		return new CollectionOptions(null, null, null, null, ValidationOptions.none());
+		return new CollectionOptions(null, null, null, null, ValidationOptions.none(), null);
+	}
+
+	/**
+	 * Quick way to set up {@link CollectionOptions} for a Time Series collection. For more advanced settings use
+	 * {@link #timeSeries(TimeSeriesOptions)}.
+	 * 
+	 * @param timeField The name of the property which contains the date in each time series document. Must not be
+	 *          {@literal null}.
+	 * @return new instance of {@link CollectionOptions}.
+	 * @see #timeSeries(TimeSeriesOptions)
+	 * @since 3.3
+	 */
+	public static CollectionOptions timeSeries(String timeField) {
+		return empty().timeSeries(TimeSeriesOptions.timeSeries(timeField));
 	}
 
 	/**
@@ -99,7 +119,7 @@ public class CollectionOptions {
 	 * @since 2.0
 	 */
 	public CollectionOptions capped() {
-		return new CollectionOptions(size, maxDocuments, true, collation, validationOptions);
+		return new CollectionOptions(size, maxDocuments, true, collation, validationOptions, null);
 	}
 
 	/**
@@ -110,7 +130,7 @@ public class CollectionOptions {
 	 * @since 2.0
 	 */
 	public CollectionOptions maxDocuments(long maxDocuments) {
-		return new CollectionOptions(size, maxDocuments, capped, collation, validationOptions);
+		return new CollectionOptions(size, maxDocuments, capped, collation, validationOptions, timeSeriesOptions);
 	}
 
 	/**
@@ -121,7 +141,7 @@ public class CollectionOptions {
 	 * @since 2.0
 	 */
 	public CollectionOptions size(long size) {
-		return new CollectionOptions(size, maxDocuments, capped, collation, validationOptions);
+		return new CollectionOptions(size, maxDocuments, capped, collation, validationOptions, timeSeriesOptions);
 	}
 
 	/**
@@ -132,7 +152,7 @@ public class CollectionOptions {
 	 * @since 2.0
 	 */
 	public CollectionOptions collation(@Nullable Collation collation) {
-		return new CollectionOptions(size, maxDocuments, capped, collation, validationOptions);
+		return new CollectionOptions(size, maxDocuments, capped, collation, validationOptions, timeSeriesOptions);
 	}
 
 	/**
@@ -252,7 +272,20 @@ public class CollectionOptions {
 	public CollectionOptions validation(ValidationOptions validationOptions) {
 
 		Assert.notNull(validationOptions, "ValidationOptions must not be null!");
-		return new CollectionOptions(size, maxDocuments, capped, collation, validationOptions);
+		return new CollectionOptions(size, maxDocuments, capped, collation, validationOptions, timeSeriesOptions);
+	}
+
+	/**
+	 * Create new {@link CollectionOptions} with the given {@link TimeSeriesOptions}.
+	 *
+	 * @param timeSeriesOptions must not be {@literal null}.
+	 * @return new instance of {@link CollectionOptions}.
+	 * @since 3.3
+	 */
+	public CollectionOptions timeSeries(TimeSeriesOptions timeSeriesOptions) {
+
+		Assert.notNull(timeSeriesOptions, "TimeSeriesOptions must not be null!");
+		return new CollectionOptions(size, maxDocuments, capped, collation, validationOptions, timeSeriesOptions);
 	}
 
 	/**
@@ -303,6 +336,16 @@ public class CollectionOptions {
 		return validationOptions.isEmpty() ? Optional.empty() : Optional.of(validationOptions);
 	}
 
+	/**
+	 * Get the {@link TimeSeriesOptions} if available.
+	 *
+	 * @return {@link Optional#empty()} if not specified.
+	 * @since 3.3
+	 */
+	public Optional<TimeSeriesOptions> getTimeSeriesOptions() {
+		return Optional.ofNullable(timeSeriesOptions);
+	}
+
 	/**
 	 * Encapsulation of ValidationOptions options.
 	 *
@@ -398,4 +441,87 @@ public class CollectionOptions {
 			return !Optionals.isAnyPresent(getValidator(), getValidationAction(), getValidationLevel());
 		}
 	}
+
+	/**
+	 * Options applicable to Time Series collections.
+	 *
+	 * @author Christoph Strobl
+	 * @since 3.3
+	 * @see <a href=
+	 *      "https://docs.mongodb.com/manual/core/timeseries-collections">https://docs.mongodb.com/manual/core/timeseries-collections</a>
+	 */
+	public static class TimeSeriesOptions {
+
+		private final String timeField;
+
+		@Nullable //
+		private String metaField;
+
+		private Granularity granularity;
+
+		private TimeSeriesOptions(String timeField, @Nullable String metaField, Granularity granularity) {
+
+			this.timeField = timeField;
+			this.metaField = metaField;
+			this.granularity = granularity;
+		}
+
+		/**
+		 * Create a new instance of {@link TimeSeriesOptions} using the given field as its {@literal timeField}. The one,
+		 * that contains the date in each time series document. <br />
+		 * {@link Field#name() Annotated fieldnames} will be considered during the mapping process.
+		 *
+		 * @param timeField must not be {@literal null}.
+		 * @return new instance of {@link TimeSeriesOptions}.
+		 */
+		public static TimeSeriesOptions timeSeries(String timeField) {
+			return new TimeSeriesOptions(timeField, null, Granularities.DEFAULT);
+		}
+
+		/**
+		 * Set the name of the field which contains metadata in each time series document. Should not be the {@literal id}
+		 * nor {@link TimeSeriesOptions#timeSeries(String)} timeField} nor point to an {@literal array} or
+		 * {@link java.util.Collection}. <br />
+		 * {@link Field#name() Annotated fieldnames} will be considered during the mapping process.
+		 *
+		 * @param metaField must not be {@literal null}.
+		 * @return new instance of {@link TimeSeriesOptions}.
+		 */
+		public TimeSeriesOptions metaField(String metaField) {
+			return new TimeSeriesOptions(timeField, metaField, granularity);
+		}
+
+		/**
+		 * Select the {@link Granularity} parameter to define how data in the time series collection is organized. Select
+		 * one that is closest to the time span between incoming measurements.
+		 *
+		 * @return new instance of {@link TimeSeriesOptions}.
+		 */
+		public TimeSeriesOptions granularity(Granularity granularity) {
+			return new TimeSeriesOptions(timeField, metaField, granularity);
+		}
+
+		/**
+		 * @return never {@literal null}.
+		 */
+		public String getTimeField() {
+			return timeField;
+		}
+
+		/**
+		 * @return can be {@literal null}. Might be an {@literal empty} {@link String} as well, so maybe check via
+		 *         {@link org.springframework.util.StringUtils#hasText(String)}.
+		 */
+		@Nullable
+		public String getMetaField() {
+			return metaField;
+		}
+
+		/**
+		 * @return never {@literal null}.
+		 */
+		public Granularity getGranularity() {
+			return granularity;
+		}
+	}
 }

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

@@ -29,19 +29,23 @@ import org.springframework.data.mapping.PersistentEntity;
 import org.springframework.data.mapping.PersistentPropertyAccessor;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
+import org.springframework.data.mongodb.core.CollectionOptions.TimeSeriesOptions;
 import org.springframework.data.mongodb.core.convert.MongoWriter;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
 import org.springframework.data.mongodb.core.mapping.MongoSimpleTypes;
+import org.springframework.data.mongodb.core.mapping.TimeSeries;
 import org.springframework.data.mongodb.core.query.Collation;
 import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.data.mongodb.core.timeseries.Granularities;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.LinkedMultiValueMap;
 import org.springframework.util.MultiValueMap;
 import org.springframework.util.ObjectUtils;
+import org.springframework.util.StringUtils;
 
 /**
  * Common operations performed on an entity in the context of it's mapping metadata.
@@ -778,6 +782,24 @@ class EntityOperations {
 		 * @return
 		 */
 		Optional<Collation> getCollation(Query query);
+
+		/**
+		 * Derive the applicable {@link CollectionOptions} for the given type.
+		 *
+		 * @return never {@literal null}.
+		 * @since 3.3
+		 */
+		CollectionOptions getCollectionOptions();
+
+		/**
+		 * Map the fields of a given {@link TimeSeriesOptions} against the target domain type to consider potentially
+		 * annotated field names.
+		 *
+		 * @param options must not be {@literal null}.
+		 * @return never {@literal null}.
+		 * @since 3.3
+		 */
+		TimeSeriesOptions mapTimeSeriesOptions(TimeSeriesOptions options);
 	}
 
 	/**
@@ -817,6 +839,16 @@ class EntityOperations {
 
 			return query.getCollation();
 		}
+
+		@Override
+		public CollectionOptions getCollectionOptions() {
+			return CollectionOptions.empty();
+		}
+
+		@Override
+		public TimeSeriesOptions mapTimeSeriesOptions(TimeSeriesOptions options) {
+			return options;
+		}
 	}
 
 	/**
@@ -854,6 +886,46 @@ class EntityOperations {
 
 			return Optional.ofNullable(entity.getCollation());
 		}
+
+		@Override
+		public CollectionOptions getCollectionOptions() {
+
+			CollectionOptions collectionOptions = CollectionOptions.empty();
+			if (entity.hasCollation()) {
+				collectionOptions = collectionOptions.collation(entity.getCollation());
+			}
+
+			if (entity.isAnnotationPresent(TimeSeries.class)) {
+
+				TimeSeries timeSeries = entity.getRequiredAnnotation(TimeSeries.class);
+				TimeSeriesOptions options = TimeSeriesOptions.timeSeries(timeSeries.timeField());
+				if (StringUtils.hasText(timeSeries.metaField())) {
+					options = options.metaField(timeSeries.metaField());
+				}
+				if (!Granularities.DEFAULT.equals(timeSeries.granularity())) {
+					options = options.granularity(timeSeries.granularity());
+				}
+				collectionOptions = collectionOptions.timeSeries(options);
+			}
+
+			return collectionOptions;
+		}
+
+		@Override
+		public TimeSeriesOptions mapTimeSeriesOptions(TimeSeriesOptions source) {
+
+			TimeSeriesOptions target = TimeSeriesOptions.timeSeries(mappedNameOrDefault(source.getTimeField()));
+
+			if (StringUtils.hasText(source.getMetaField())) {
+				target = target.metaField(mappedNameOrDefault(source.getMetaField()));
+			}
+			return target.granularity(source.getGranularity());
+		}
+
+		private String mappedNameOrDefault(String name) {
+			MongoPersistentProperty persistentProperty = entity.getPersistentProperty(name);
+			return persistentProperty != null ? persistentProperty.getFieldName() : name;
+		}
 	}
 
 }

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

@@ -99,6 +99,7 @@ import org.springframework.data.mongodb.core.query.NearQuery;
 import org.springframework.data.mongodb.core.query.Query;
 import org.springframework.data.mongodb.core.query.UpdateDefinition;
 import org.springframework.data.mongodb.core.query.UpdateDefinition.ArrayFilter;
+import org.springframework.data.mongodb.core.timeseries.Granularities;
 import org.springframework.data.mongodb.core.validation.Validator;
 import org.springframework.data.mongodb.util.BsonUtils;
 import org.springframework.data.projection.SpelAwareProxyProjectionFactory;
@@ -597,7 +598,7 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 	 * @see org.springframework.data.mongodb.core.MongoOperations#createCollection(java.lang.Class)
 	 */
 	public <T> MongoCollection<Document> createCollection(Class<T> entityClass) {
-		return createCollection(entityClass, CollectionOptions.empty());
+		return createCollection(entityClass, operations.forType(entityClass).getCollectionOptions());
 	}
 
 	/*
@@ -2435,6 +2436,19 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 				co.validationOptions(options);
 			}
 
+			if(collectionOptions.containsKey("timeseries")) {
+
+				Document timeSeries = collectionOptions.get("timeseries", Document.class);
+				com.mongodb.client.model.TimeSeriesOptions options = new com.mongodb.client.model.TimeSeriesOptions(timeSeries.getString("timeField"));
+				if(timeSeries.containsKey("metaField")) {
+					options.metaField(timeSeries.getString("metaField"));
+				}
+				if(timeSeries.containsKey("granularity")) {
+					options.granularity(TimeSeriesGranularity.valueOf(timeSeries.getString("granularity").toUpperCase()));
+				}
+				co.timeSeriesOptions(options);
+			}
+
 			db.createCollection(collectionName, co);
 
 			MongoCollection<Document> coll = db.getCollection(collectionName, Document.class);
@@ -2589,6 +2603,18 @@ public class MongoTemplate implements MongoOperations, ApplicationContextAware,
 
 			collectionOptions.getValidationOptions().ifPresent(it -> it.getValidator() //
 					.ifPresent(val -> doc.put("validator", getMappedValidator(val, targetType))));
+
+			collectionOptions.getTimeSeriesOptions().map(operations.forType(targetType)::mapTimeSeriesOptions).ifPresent(it -> {
+
+				Document timeseries = new Document("timeField", it.getTimeField());
+				if(StringUtils.hasText(it.getMetaField())) {
+					timeseries.append("metaField", it.getMetaField());
+				}
+				if(!Granularities.DEFAULT.equals(it.getGranularity())) {
+					timeseries.append("granularity", it.getGranularity().name().toLowerCase());
+				}
+				doc.put("timeseries", timeseries);
+			});
 		}
 
 		return doc;

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

@@ -17,6 +17,7 @@ package org.springframework.data.mongodb.core;
 
 import static org.springframework.data.mongodb.core.query.SerializationUtils.*;
 
+import org.springframework.data.mongodb.core.timeseries.Granularities;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.util.function.Tuple2;
@@ -665,7 +666,7 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 	 * @see org.springframework.data.mongodb.core.ReactiveMongoOperations#createCollection(java.lang.Class)
 	 */
 	public <T> Mono<MongoCollection<Document>> createCollection(Class<T> entityClass) {
-		return createCollection(entityClass, CollectionOptions.empty());
+		return createCollection(entityClass, operations.forType(entityClass).getCollectionOptions());
 	}
 
 	/*
@@ -2505,6 +2506,20 @@ public class ReactiveMongoTemplate implements ReactiveMongoOperations, Applicati
 			result.validationOptions(validationOptions);
 		});
 
+		collectionOptions.getTimeSeriesOptions().map(operations.forType(entityType)::mapTimeSeriesOptions).ifPresent(it -> {
+
+			TimeSeriesOptions options = new TimeSeriesOptions(it.getTimeField());
+
+			if(StringUtils.hasText(it.getMetaField())) {
+				options.metaField(it.getMetaField());
+			}
+			if(!Granularities.DEFAULT.equals(it.getGranularity())) {
+				options.granularity(TimeSeriesGranularity.valueOf(it.getGranularity().name().toUpperCase()));
+			}
+
+			result.timeSeriesOptions(options);
+		});
+
 		return result;
 	}
 

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

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2021 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.mapping;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Inherited;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.core.annotation.AliasFor;
+import org.springframework.data.mongodb.core.timeseries.Granularities;
+
+/**
+ * Identifies a domain object to be persisted to a MongoDB Time Series collection.
+ *
+ * @author Christoph Strobl
+ * @since 3.3
+ * @see <a href=
+ *      "https://docs.mongodb.com/manual/core/timeseries-collections">https://docs.mongodb.com/manual/core/timeseries-collections</a>
+ */
+@Inherited
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ ElementType.TYPE })
+@Document
+public @interface TimeSeries {
+
+	/**
+	 * The collection the document representing the entity is supposed to be stored in. If not configured, a default
+	 * collection name will be derived from the type's name. The attribute supports SpEL expressions to dynamically
+	 * calculate the collection based on a per operation basis.
+	 *
+	 * @return the name of the collection to be used.
+	 * @see Document#collection()
+	 */
+	@AliasFor(annotation = Document.class, attribute = "collection")
+	String collection() default "";
+
+	/**
+	 * The name of the property which contains the date in each time series document. <br />
+	 * {@link Field#name() Annotated fieldnames} will be considered during the mapping process.
+	 *
+	 * @return never {@literal null}.
+	 */
+	String timeField();
+
+	/**
+	 * The name of the field which contains metadata in each time series document. Should not be the {@literal id} nor
+	 * {@link #timeField()} nor point to an {@literal array} or {@link java.util.Collection}. <br />
+	 * {@link Field#name() Annotated fieldnames} will be considered during the mapping process.
+	 *
+	 * @return empty {@link String} by default.
+	 */
+	String metaField() default "";
+
+	/**
+	 * Select the {@link Granularities granularity} parameter to define how data in the time series collection is
+	 * organized.
+	 *
+	 * @return {@link Granularities#DEFAULT server default} by default.
+	 */
+	Granularities granularity() default Granularities.DEFAULT;
+
+	/**
+	 * Defines the collation to apply when executing a query or creating indexes.
+	 *
+	 * @return an empty {@link String} by default.
+	 * @see Document#collation()
+	 */
+	@AliasFor(annotation = Document.class, attribute = "collation")
+	String collation() default "";
+
+}

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

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2021 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.timeseries;
+
+/**
+ * {@link Granularity Granularities} available for Time Series data.
+ *
+ * @author Christoph Strobl
+ * @since 3.3
+ */
+public enum Granularities implements Granularity {
+
+	/**
+	 * Server default value to indicate no explicit value should be sent.
+	 */
+	DEFAULT,
+
+	/**
+	 * High frequency ingestion.
+	 */
+	SECONDS,
+
+	/**
+	 * Medium frequency ingestion.
+	 */
+	MINUTES,
+
+	/**
+	 * Low frequency ingestion.
+	 */
+	HOURS
+}

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

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2021 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.timeseries;
+
+/**
+ * The Granularity of time series data that is closest to the time span between incoming measurements.
+ *
+ * @author Christoph Strobl
+ * @since 3.3
+ */
+public interface Granularity {
+
+	String name();
+}

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

@@ -19,12 +19,14 @@ import static org.mockito.Mockito.*;
 import static org.springframework.data.mongodb.core.aggregation.Aggregation.*;
 import static org.springframework.data.mongodb.test.util.Assertions.*;
 
+import com.mongodb.client.model.*;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
 import java.math.BigInteger;
 import java.time.Duration;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -82,6 +84,7 @@ import org.springframework.data.mongodb.core.index.MongoPersistentEntityIndexCre
 import org.springframework.data.mongodb.core.mapping.Field;
 import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
 import org.springframework.data.mongodb.core.mapping.Sharded;
+import org.springframework.data.mongodb.core.mapping.TimeSeries;
 import org.springframework.data.mongodb.core.mapping.event.AbstractMongoEventListener;
 import org.springframework.data.mongodb.core.mapping.event.AfterConvertCallback;
 import org.springframework.data.mongodb.core.mapping.event.AfterSaveCallback;
@@ -98,6 +101,7 @@ 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.core.timeseries.Granularities;
 import org.springframework.lang.Nullable;
 import org.springframework.test.util.ReflectionTestUtils;
 import org.springframework.util.CollectionUtils;
@@ -117,15 +121,6 @@ import com.mongodb.client.MongoClient;
 import com.mongodb.client.MongoCollection;
 import com.mongodb.client.MongoCursor;
 import com.mongodb.client.MongoDatabase;
-import com.mongodb.client.model.CountOptions;
-import com.mongodb.client.model.CreateCollectionOptions;
-import com.mongodb.client.model.DeleteOptions;
-import com.mongodb.client.model.FindOneAndDeleteOptions;
-import com.mongodb.client.model.FindOneAndReplaceOptions;
-import com.mongodb.client.model.FindOneAndUpdateOptions;
-import com.mongodb.client.model.MapReduceAction;
-import com.mongodb.client.model.ReplaceOptions;
-import com.mongodb.client.model.UpdateOptions;
 import com.mongodb.client.result.DeleteResult;
 import com.mongodb.client.result.UpdateResult;
 
@@ -2256,6 +2251,30 @@ public class MongoTemplateUnitTests extends MongoOperationsUnitTests {
 				.isThrownBy(() -> template.save(new ArrayList<>(Arrays.asList(1, 2, 3)), "myList"));
 	}
 
+	@Test // GH-3731
+	void createCollectionShouldSetUpTimeSeriesWithDefaults() {
+
+		template.createCollection(TimeSeriesTypeWithDefaults.class);
+
+		ArgumentCaptor<CreateCollectionOptions> options = ArgumentCaptor.forClass(CreateCollectionOptions.class);
+		verify(db).createCollection(any(), options.capture());
+
+		assertThat(options.getValue().getTimeSeriesOptions().toString())
+				.isEqualTo(new com.mongodb.client.model.TimeSeriesOptions("timestamp").toString());
+	}
+
+	@Test // GH-3731
+	void createCollectionShouldSetUpTimeSeries() {
+
+		template.createCollection(TimeSeriesType.class);
+
+		ArgumentCaptor<CreateCollectionOptions> options = ArgumentCaptor.forClass(CreateCollectionOptions.class);
+		verify(db).createCollection(any(), options.capture());
+
+		assertThat(options.getValue().getTimeSeriesOptions().toString())
+				.isEqualTo(new com.mongodb.client.model.TimeSeriesOptions("time_stamp").metaField("meta").granularity(TimeSeriesGranularity.HOURS).toString());
+	}
+
 	class AutogenerateableId {
 
 		@Id BigInteger id;
@@ -2358,6 +2377,23 @@ public class MongoTemplateUnitTests extends MongoOperationsUnitTests {
 		WithNamedFields nested;
 	}
 
+	@TimeSeries(timeField = "timestamp")
+	static class TimeSeriesTypeWithDefaults {
+
+		String id;
+		Instant timestamp;
+	}
+
+	@TimeSeries(timeField = "timestamp", metaField = "meta", granularity = Granularities.HOURS)
+	static class TimeSeriesType {
+
+		String id;
+
+		@Field("time_stamp")
+		Instant timestamp;
+		Object meta;
+	}
+
 	static class TypeImplementingIterator implements Iterator {
 
 		@Override

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

@@ -20,15 +20,21 @@ import static org.mockito.Mockito.*;
 import static org.springframework.data.mongodb.core.aggregation.Aggregation.*;
 import static org.springframework.data.mongodb.test.util.Assertions.assertThat;
 
+import com.mongodb.client.model.TimeSeriesGranularity;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
+import org.springframework.data.mongodb.core.MongoTemplateUnitTests.TimeSeriesType;
+import org.springframework.data.mongodb.core.MongoTemplateUnitTests.TimeSeriesTypeWithDefaults;
 import org.springframework.data.mongodb.core.convert.MongoCustomConversions;
+import org.springframework.data.mongodb.core.mapping.TimeSeries;
+import org.springframework.data.mongodb.core.timeseries.Granularities;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import java.time.Duration;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -1426,6 +1432,30 @@ public class ReactiveMongoTemplateUnitTests {
 				.isThrownBy(() -> template.insert(publisher));
 	}
 
+	@Test // GH-3731
+	void createCollectionShouldSetUpTimeSeriesWithDefaults() {
+
+		template.createCollection(TimeSeriesTypeWithDefaults.class).subscribe();
+
+		ArgumentCaptor<CreateCollectionOptions> options = ArgumentCaptor.forClass(CreateCollectionOptions.class);
+		verify(db).createCollection(any(), options.capture());
+
+		assertThat(options.getValue().getTimeSeriesOptions().toString())
+				.isEqualTo(new com.mongodb.client.model.TimeSeriesOptions("timestamp").toString());
+	}
+
+	@Test // GH-3731
+	void createCollectionShouldSetUpTimeSeries() {
+
+		template.createCollection(TimeSeriesType.class).subscribe();
+
+		ArgumentCaptor<CreateCollectionOptions> options = ArgumentCaptor.forClass(CreateCollectionOptions.class);
+		verify(db).createCollection(any(), options.capture());
+
+		assertThat(options.getValue().getTimeSeriesOptions().toString())
+				.isEqualTo(new com.mongodb.client.model.TimeSeriesOptions("time_stamp").metaField("meta").granularity(TimeSeriesGranularity.HOURS).toString());
+	}
+
 	private void stubFindSubscribe(Document document) {
 
 		Publisher<Document> realPublisher = Flux.just(document);
@@ -1483,6 +1513,23 @@ public class ReactiveMongoTemplateUnitTests {
 		List<Integer> grades;
 	}
 
+	@TimeSeries(timeField = "timestamp")
+	static class TimeSeriesTypeWithDefaults {
+
+		String id;
+		Instant timestamp;
+	}
+
+	@TimeSeries(timeField = "timestamp", metaField = "meta", granularity = Granularities.HOURS)
+	static class TimeSeriesType {
+
+		String id;
+
+		@Field("time_stamp")
+		Instant timestamp;
+		Object meta;
+	}
+
 	static class ValueCapturingEntityCallback<T> {
 
 		private final List<T> values = new ArrayList<>(1);

src/main/asciidoc/new-features.adoc

@@ -5,6 +5,7 @@
 == What's New in Spring Data MongoDB 3.3
 
 * Extended support for <<mapping-usage.document-references, referencing>> entities.
+* Support for <<time-series, Time Series>> collections.
 * Include/exclude `null` properties on write to `Document` through `@Field(write=…)`.
 * Support for <<mapping-usage-indexes.wildcard-index>>.
 

src/main/asciidoc/reference/mongodb.adoc

@@ -3382,3 +3382,4 @@ class GridFsClient {
 
 include::tailable-cursors.adoc[]
 include::change-streams.adoc[]
+include::time-series.adoc[]

src/main/asciidoc/reference/time-series.adoc

@@ -0,0 +1,45 @@
+[[time-series]]
+== Time Series
+
+MongoDB 5.0 introduced https://docs.mongodb.com/manual/core/timeseries-collections/[Time Series] collections optimized to efficiently store sequences of measurements.
+Those collections need to be actively created before inserting any data. This can be done by manually executing the command, defining time series collection options or extracting options from a `@TimeSeries` annotation as shown in the examples below.
+
+.Create a Time Series Collection
+====
+.Create a Time Series via the MongoDB Driver
+[code, java]
+----
+template.execute(db -> {
+
+    com.mongodb.client.model.CreateCollectionOptions options = new CreateCollectionOptions();
+    options.timeSeriesOptions(new TimeSeriesOptions("timestamp"));
+
+    db.createCollection("weather", options);
+    return "OK";
+});
+----
+
+.Create a Time Series Collection with CollectionOptions
+[code, java]
+----
+template.createCollection("weather", CollectionOptions.timeSeries("timestamp"));
+----
+
+.Create a Time Series Collection derived from an Annotation
+[code, java]
+----
+@TimeSeries(collection="weather", timeField = "timestamp")
+public class Measurement {
+
+    String id;
+    Instant timestamp;
+    // ...
+}
+
+template.createCollection(Measurement.class);
+----
+====
+
+The snippets above can easily be transferred to the reactive API offering the very same methods.
+Just make sure to _subscribe_.
+