Add support for $top & $topN aggregation operators.

Closes #4139
Original pull request: #4182.

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

@@ -123,6 +123,109 @@ public class SelectionOperators {
 		}
 	}
 
+	/**
+	 * {@link AbstractAggregationExpression} to return the top element according to the specified {@link #sortBy(Sort)
+	 * order}.
+	 */
+	public static class Top extends AbstractAggregationExpression {
+
+		private Top(Object value) {
+			super(value);
+		}
+
+		/**
+		 * In case a limit value ({@literal n}) is present {@literal $topN} is used instead of {@literal $top}.
+		 *
+		 * @return
+		 */
+		@Override
+		protected String getMongoMethod() {
+			return get("n") == null ? "$top" : "$topN";
+		}
+
+		/**
+		 * @return new instance of {@link Top}.
+		 */
+		public static Top top() {
+			return new Top(Collections.emptyMap());
+		}
+
+		/**
+		 * @param numberOfResults Limits the number of returned elements to the given value.
+		 * @return new instance of {@link Top}.
+		 */
+		public static Top top(int numberOfResults) {
+			return top().limit(numberOfResults);
+		}
+
+		/**
+		 * Limits the number of returned elements to the given value.
+		 *
+		 * @param numberOfResults
+		 * @return new instance of {@link Top}.
+		 */
+		public Top limit(int numberOfResults) {
+			return limit((Object) numberOfResults);
+		}
+
+		/**
+		 * Limits the number of returned elements to the value defined by the given {@link AggregationExpression
+		 * expression}.
+		 *
+		 * @param expression must not be {@literal null}.
+		 * @return new instance of {@link Top}.
+		 */
+		public Top limit(AggregationExpression expression) {
+			return limit((Object) expression);
+		}
+
+		private Top limit(Object value) {
+			return new Top(append("n", value));
+		}
+
+		/**
+		 * Define result ordering.
+		 *
+		 * @param sort must not be {@literal null}.
+		 * @return new instance of {@link Top}.
+		 */
+		public Top sortBy(Sort sort) {
+			return new Top(append("sortBy", sort));
+		}
+
+		/**
+		 * Define result ordering.
+		 *
+		 * @param out must not be {@literal null}.
+		 * @return new instance of {@link Top}.
+		 */
+		public Top output(Fields out) {
+			return new Top(append("output", out));
+		}
+
+		/**
+		 * Define fields included in the output for each element.
+		 *
+		 * @param fieldNames must not be {@literal null}.
+		 * @return new instance of {@link Top}.
+		 * @see #output(Fields)
+		 */
+		public Top output(String... fieldNames) {
+			return output(Fields.fields(fieldNames));
+		}
+
+		/**
+		 * Define expressions building the value included in the output for each element.
+		 *
+		 * @param out must not be {@literal null}.
+		 * @return new instance of {@link Top}.
+		 * @see #output(Fields)
+		 */
+		public Top output(AggregationExpression... out) {
+			return new Top(append("output", Arrays.asList(out)));
+		}
+	}
+
 	/**
 	 * {@link AbstractAggregationExpression} to return the {@literal $firstN} elements.
 	 */

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

@@ -220,6 +220,10 @@ public class MethodReferenceNode extends ExpressionNode {
 				.mappingParametersTo("n", "input"));
 		map.put("lastN", mapArgRef().forOperator("$lastN") //
 				.mappingParametersTo("n", "input"));
+		map.put("top", mapArgRef().forOperator("$top") //
+				.mappingParametersTo("output", "sortBy"));
+		map.put("topN", mapArgRef().forOperator("$topN") //
+				.mappingParametersTo("n", "output", "sortBy"));
 
 		// CONVERT OPERATORS
 		map.put("convert", mapArgRef().forOperator("$convert") //

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

@@ -89,6 +89,46 @@ class SelectionOperatorUnitTests {
 				"""));
 	}
 
+	@Test // GH-4139
+	void topMapsFieldNamesCorrectly() {
+
+		MongoMappingContext mappingContext = new MongoMappingContext();
+		RelaxedTypeBasedAggregationOperationContext aggregationContext = new RelaxedTypeBasedAggregationOperationContext(
+				Player.class, mappingContext,
+				new QueryMapper(new MappingMongoConverter(NoOpDbRefResolver.INSTANCE, mappingContext)));
+
+		Document document = SelectionOperators.Top.top().output(Fields.fields("playerId", "score"))
+				.sortBy(Sort.by(Direction.DESC, "score")).toDocument(aggregationContext);
+
+		assertThat(document).isEqualTo(Document.parse("""
+				{
+				   $top:
+				   {
+				      output: [ "$player_id", "$s_cor_e" ],
+				      sortBy: { "s_cor_e": -1 }
+				   }
+				}
+				"""));
+	}
+
+	@Test // GH-4139
+	void topNRenderedCorrectly() {
+
+		Document document = SelectionOperators.Top.top().output(Fields.fields("playerId", "score"))
+				.sortBy(Sort.by(Direction.DESC, "score")).limit(3).toDocument(Aggregation.DEFAULT_CONTEXT);
+
+		assertThat(document).isEqualTo(Document.parse("""
+				{
+				   $topN:
+				   {
+				      n : 3,
+				      output: [ "$playerId", "$score" ],
+				      sortBy: { "score": -1 }
+				   }
+				}
+				"""));
+	}
+
 	@Test // GH-4139
 	void firstNMapsFieldNamesCorrectly() {
 

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

@@ -1184,6 +1184,16 @@ public class SpelExpressionTransformerUnitTests {
 		assertThat(transform("bottomN(3, new String[]{\"$playerId\", \"$score\" }, { \"score\" : -1 })")).isEqualTo("{ $bottomN : { n : 3, output: [ \"$playerId\", \"$score\" ], sortBy: { \"score\": -1 }}}");
 	}
 
+	@Test // GH-4139
+	void shouldRenderTop() {
+		assertThat(transform("top(new String[]{\"$playerId\", \"$score\" }, { \"score\" : -1 })")).isEqualTo("{ $top : { output: [ \"$playerId\", \"$score\" ], sortBy: { \"score\": -1 }}}");
+	}
+
+	@Test // GH-4139
+	void shouldRenderTopN() {
+		assertThat(transform("topN(3, new String[]{\"$playerId\", \"$score\" }, { \"score\" : -1 })")).isEqualTo("{ $topN : { n : 3, output: [ \"$playerId\", \"$score\" ], sortBy: { \"score\": -1 }}}");
+	}
+
 	@Test // GH-4139
 	void shouldRenderFirstN() {
 		assertThat(transform("firstN(3, \"$score\")")).isEqualTo("{ $firstN : { n : 3, input : \"$score\" }}");