Accept expression as input for filter aggregation operator

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-mongodb-parent</artifactId>
-	<version>4.2.0-SNAPSHOT</version>
+	<version>4.2.x-4394-SNAPSHOT</version>
 	<packaging>pom</packaging>
 
 	<name>Spring Data MongoDB</name>

spring-data-mongodb-benchmarks/pom.xml

@@ -7,7 +7,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>4.2.0-SNAPSHOT</version>
+		<version>4.2.x-4394-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb-distribution/pom.xml

@@ -15,7 +15,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>4.2.0-SNAPSHOT</version>
+		<version>4.2.x-4394-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb/pom.xml

@@ -13,7 +13,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>4.2.0-SNAPSHOT</version>
+		<version>4.2.x-4394-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

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

@@ -214,6 +214,10 @@ public AsBuilder filter() {
 				return Filter.filter(fieldReference);
 			}
 
+			if(usesExpression()) {
+				return Filter.filter(expression);
+			}
+
 			Assert.state(values != null, "Values must not be null");
 			return Filter.filter(new ArrayList<>(values));
 		}
@@ -649,6 +653,19 @@ public static AsBuilder filter(Field field) {
 			return new FilterExpressionBuilder().filter(field);
 		}
 
+		/**
+		 * Set the {@link AggregationExpression} resolving to an arry to apply the {@code $filter} to.
+		 *
+		 * @param expression must not be {@literal null}.
+		 * @return never {@literal null}.
+		 * @since 4.2
+		 */
+		public static AsBuilder filter(AggregationExpression expression) {
+
+			Assert.notNull(expression, "Field must not be null");
+			return new FilterExpressionBuilder().filter(expression);
+		}
+
 		/**
 		 * Set the {@literal values} to apply the {@code $filter} to.
 		 *
@@ -681,7 +698,13 @@ private Document toFilter(ExposedFields exposedFields, AggregationOperationConte
 		}
 
 		private Object getMappedInput(AggregationOperationContext context) {
-			return input instanceof Field field ? context.getReference(field).toString() : input;
+			if(input instanceof Field field) {
+				return context.getReference(field).toString();
+			}
+			if(input instanceof AggregationExpression expression) {
+				return expression.toDocument(context);
+			}
+			return input;
 		}
 
 		private Object getMappedCondition(AggregationOperationContext context) {
@@ -715,6 +738,15 @@ public interface InputBuilder {
 			 * @return
 			 */
 			AsBuilder filter(Field field);
+
+			/**
+			 * Set the {@link AggregationExpression} resolving to an array to apply the {@code $filter} to.
+			 *
+			 * @param expression must not be {@literal null}.
+			 * @return
+			 * @since 4.2
+			 */
+			AsBuilder filter(AggregationExpression expression);
 		}
 
 		/**
@@ -797,6 +829,14 @@ public AsBuilder filter(Field field) {
 				return this;
 			}
 
+			@Override
+			public AsBuilder filter(AggregationExpression expression) {
+
+				Assert.notNull(expression, "Expression must not be null");
+				filter.input = expression;
+				return this;
+			}
+
 			@Override
 			public ConditionBuilder as(String variableName) {
 

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

@@ -117,6 +117,23 @@ void shouldConstructFilterExpressionCorrectlyWhenConditionContainsFieldReference
 		assertThat($filter).isEqualTo(new Document(expected));
 	}
 
+	@Test // GH-4394
+	void filterShouldAcceptExpression() {
+
+		Document $filter = ArrayOperators.arrayOf(ObjectOperators.valueOf("data.metadata").toArray()).filter().as("item")
+				.by(ComparisonOperators.valueOf("item.price").greaterThan("field-1")).toDocument(Aggregation.DEFAULT_CONTEXT);
+
+		Document expected = Document.parse("""
+				{ $filter : {
+					input: { $objectToArray: "$data.metadata" },
+					as: "item",
+					cond: { $gt: [ "$$item.price", "$field-1" ] }
+				}}
+				""");
+
+		assertThat($filter).isEqualTo(expected);
+	}
+
 	private Document extractFilterOperatorFromDocument(Document source) {
 
 		List<Object> pipeline = DocumentTestUtils.getAsDBList(source, "pipeline");