DATAMONGO-1491 - Add support for $filter (aggregation).

We new support $filter in aggregation pipeline.

Aggregation.newAggregation(Sales.class,
	Aggregation.project()
		.and(filter("items").as("item").by(GTE.of(field("item.price"), 100)))
		.as("items"))

Original pull request: #412.

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

@@ -26,6 +26,7 @@ import org.springframework.data.domain.Sort;
 import org.springframework.data.domain.Sort.Direction;
 import org.springframework.data.mongodb.core.aggregation.ExposedFields.ExposedField;
 import org.springframework.data.mongodb.core.aggregation.ExposedFields.FieldReference;
+import org.springframework.data.mongodb.core.aggregation.ExposedFields.DirectFieldReference;
 import org.springframework.data.mongodb.core.aggregation.Fields.AggregationField;
 import org.springframework.data.mongodb.core.aggregation.FieldsExposingAggregationOperation.InheritsFieldsAggregationOperation;
 import org.springframework.data.mongodb.core.query.Criteria;
@@ -555,7 +556,7 @@ public class Aggregation {
 		 */
 		@Override
 		public FieldReference getReference(Field field) {
-			return new FieldReference(new ExposedField(field, true));
+			return new DirectFieldReference(new ExposedField(field, true));
 		}
 
 		/* 
@@ -564,7 +565,7 @@ public class Aggregation {
 		 */
 		@Override
 		public FieldReference getReference(String name) {
-			return new FieldReference(new ExposedField(new AggregationField(name), true));
+			return new DirectFieldReference(new ExposedField(new AggregationField(name), true));
 		}
 	}
 

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

@@ -0,0 +1,294 @@
+/*
+ * Copyright 2016. 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
+ *
+ *      http://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.aggregation;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.bson.Document;
+import org.springframework.data.mongodb.core.aggregation.ExposedFields.ExposedField;
+import org.springframework.data.mongodb.core.aggregation.ExposedFields.FieldReference;
+import org.springframework.util.Assert;
+
+import com.mongodb.BasicDBObject;
+import com.mongodb.DBObject;
+
+/**
+ * @author Christoph Strobl
+ * @since 1.10
+ */
+public interface AggregationExpressions {
+
+	/**
+	 * {@code $filter} {@link AggregationExpression} allows to select a subset of the array to return based on the
+	 * specified condition.
+	 *
+	 * @author Christoph Strobl
+	 * @since 1.10
+	 */
+	class Filter implements AggregationExpression {
+
+		private Object input;
+		private ExposedField as;
+		private Object condition;
+
+		private Filter() {
+			// used by builder
+		}
+
+		/**
+		 * Set the {@literal field} to apply the {@code $filter} to.
+		 *
+		 * @param field must not be {@literal null}.
+		 * @return never {@literal null}.
+		 */
+		public static AsBuilder filter(String field) {
+
+			Assert.notNull(field, "Field must not be null!");
+			return filter(Fields.field(field));
+		}
+
+		/**
+		 * Set the {@literal field} to apply the {@code $filter} to.
+		 *
+		 * @param field must not be {@literal null}.
+		 * @return never {@literal null}.
+		 */
+		public static AsBuilder filter(Field field) {
+
+			Assert.notNull(field, "Field must not be null!");
+			return new FilterExpressionBuilder().filter(field);
+		}
+
+		/**
+		 * Set the {@literal values} to apply the {@code $filter} to.
+		 *
+		 * @param values must not be {@literal null}.
+		 * @return
+		 */
+		public static AsBuilder filter(List<?> values) {
+
+			Assert.notNull(values, "Values must not be null!");
+			return new FilterExpressionBuilder().filter(values);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.aggregation.AggregationExpression#toDocument(org.springframework.data.mongodb.core.aggregation.AggregationOperationContext)
+		 */
+		@Override
+		public Document toDocument(final AggregationOperationContext context) {
+
+			return toFilter(new ExposedFieldsAggregationOperationContext(ExposedFields.from(as), context) {
+
+				@Override
+				public FieldReference getReference(Field field) {
+
+					FieldReference ref = null;
+					try {
+						ref = context.getReference(field);
+					} catch (Exception e) {
+						// just ignore that one.
+					}
+					return ref != null ? ref : super.getReference(field);
+				}
+			});
+		}
+
+		private Document toFilter(AggregationOperationContext context) {
+
+			Document filterExpression = new Document();
+
+			filterExpression.putAll(context.getMappedObject(new Document("input", getMappedInput(context))));
+			filterExpression.put("as", as.getTarget());
+
+			filterExpression.putAll(context.getMappedObject(new Document("cond", getMappedCondition(context))));
+
+			return new Document("$filter", filterExpression);
+		}
+
+		private Object getMappedInput(AggregationOperationContext context) {
+			return input instanceof Field ? context.getReference((Field) input).toString() : input;
+		}
+
+		private Object getMappedCondition(AggregationOperationContext context) {
+
+			if (!(condition instanceof AggregationExpression)) {
+				return condition;
+			}
+
+			NestedDelegatingExpressionAggregationOperationContext nea = new NestedDelegatingExpressionAggregationOperationContext(context);
+			Document mappedCondition = ((AggregationExpression) condition).toDocument(nea);
+			return mappedCondition;
+		}
+
+		/**
+		 * @author Christoph Strobl
+		 */
+		public interface InputBuilder {
+
+			/**
+			 * Set the {@literal values} to apply the {@code $filter} to.
+			 *
+			 * @param array must not be {@literal null}.
+			 * @return
+			 */
+			AsBuilder filter(List<?> array);
+
+			/**
+			 * Set the {@literal field} holding an array to apply the {@code $filter} to.
+			 *
+			 * @param field must not be {@literal null}.
+			 * @return
+			 */
+			AsBuilder filter(Field field);
+		}
+
+		/**
+		 * @author Christoph Strobl
+		 */
+		public interface AsBuilder {
+
+			/**
+			 * Set the {@literal variableName} for the elements in the input array.
+			 *
+			 * @param variableName must not be {@literal null}.
+			 * @return
+			 */
+			ConditionBuilder as(String variableName);
+		}
+
+		/**
+		 * @author Christoph Strobl
+		 */
+		public interface ConditionBuilder {
+
+			/**
+			 * Set the {@link AggregationExpression} that determines whether to include the element in the resulting array.
+			 *
+			 * @param expression must not be {@literal null}.
+			 * @return
+			 */
+			Filter by(AggregationExpression expression);
+
+			/**
+			 * Set the {@literal expression} that determines whether to include the element in the resulting array.
+			 *
+			 * @param expression must not be {@literal null}.
+			 * @return
+			 */
+			Filter by(String expression);
+
+			/**
+			 * Set the {@literal expression} that determines whether to include the element in the resulting array.
+			 *
+			 * @param expression must not be {@literal null}.
+			 * @return
+			 */
+			Filter by(Document expression);
+		}
+
+		/**
+		 * @author Christoph Strobl
+		 */
+		static final class FilterExpressionBuilder implements InputBuilder, AsBuilder, ConditionBuilder {
+
+			private final Filter filter;
+
+			FilterExpressionBuilder() {
+				this.filter = new Filter();
+			}
+
+			public static InputBuilder newBuilder() {
+				return new FilterExpressionBuilder();
+			}
+
+			/*
+			 * (non-Javadoc)
+			 * @see org.springframework.data.mongodb.core.aggregation.AggregationExpressions.Filter.InputBuilder#filter(java.util.List)
+			 */
+			@Override
+			public AsBuilder filter(List<?> array) {
+
+				Assert.notNull(array, "Array must not be null!");
+				filter.input = new ArrayList(array);
+				return this;
+			}
+
+			/*
+			 * (non-Javadoc)
+			 * @see org.springframework.data.mongodb.core.aggregation.AggregationExpressions.Filter.InputBuilder#filter(org.springframework.data.mongodb.core.aggregation.Field)
+			 */
+			@Override
+			public AsBuilder filter(Field field) {
+
+				Assert.notNull(field, "Field must not be null!");
+				filter.input = field;
+				return this;
+			}
+
+			/*
+			 * (non-Javadoc)
+			 * @see org.springframework.data.mongodb.core.aggregation.AggregationExpressions.Filter.AsBuilder#as(java.lang.String)
+			 */
+			@Override
+			public ConditionBuilder as(String variableName) {
+
+				Assert.notNull(variableName, "Variable name  must not be null!");
+				filter.as = new ExposedField(variableName, true);
+				return this;
+			}
+
+			/*
+			 * (non-Javadoc)
+			 * @see org.springframework.data.mongodb.core.aggregation.AggregationExpressions.Filter.ConditionBuilder#by(org.springframework.data.mongodb.core.aggregation.AggregationExpression)
+			 */
+			@Override
+			public Filter by(AggregationExpression condition) {
+
+				Assert.notNull(condition, "Condition must not be null!");
+				filter.condition = condition;
+				return filter;
+			}
+
+			/*
+			 * (non-Javadoc)
+			 * @see org.springframework.data.mongodb.core.aggregation.AggregationExpressions.Filter.ConditionBuilder#by(java.lang.String)
+			 */
+			@Override
+			public Filter by(String expression) {
+
+				Assert.notNull(expression, "Expression must not be null!");
+				filter.condition = expression;
+				return filter;
+			}
+
+			/*
+			 * (non-Javadoc)
+			 * @see org.springframework.data.mongodb.core.aggregation.AggregationExpressions.Filter.ConditionBuilder#by(org.bson.Document)
+			 */
+			@Override
+			public Filter by(Document expression) {
+
+				Assert.notNull(expression, "Expression must not be null!");
+				filter.condition = expression;
+				return filter;
+			}
+		}
+
+	}
+
+}

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

@@ -33,7 +33,7 @@ import org.springframework.util.Assert;
  */
 public enum AggregationFunctionExpressions {
 
-	SIZE;
+	SIZE, GTE;
 
 	/**
 	 * Returns an {@link AggregationExpression} build from the current {@link Enum} name and the given parameters.

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

@@ -24,6 +24,7 @@ import java.util.List;
 import org.springframework.data.mongodb.core.aggregation.ExposedFields.ExposedField;
 import org.springframework.util.Assert;
 import org.springframework.util.CompositeIterator;
+import org.springframework.util.ObjectUtils;
 
 /**
  * Value object to capture the fields exposed by an {@link AggregationOperation}.
@@ -104,7 +105,7 @@ public final class ExposedFields implements Iterable<ExposedField> {
 			result.add(new ExposedField(field, synthetic));
 		}
 
-		return ExposedFields.from(result);
+		return from(result);
 	}
 
 	/**
@@ -336,12 +337,36 @@ public final class ExposedFields implements Iterable<ExposedField> {
 		}
 	}
 
+	/**
+	 * A reference to an {@link ExposedField}.
+	 *
+	 * @author Christoph Strobl
+	 * @since 1.10
+	 */
+	interface FieldReference {
+
+		/**
+		 * Returns the raw, unqualified reference, i.e. the field reference without a {@literal $} prefix.
+		 *
+		 * @return
+		 */
+		String getRaw();
+
+		/**
+		 * Returns the reference value for the given field reference. Will return 1 for a synthetic, unaliased field or the
+		 * raw rendering of the reference otherwise.
+		 *
+		 * @return
+		 */
+		Object getReferenceValue();
+	}
+
 	/**
 	 * A reference to an {@link ExposedField}.
 	 * 
 	 * @author Oliver Gierke
 	 */
-	static class FieldReference {
+	static class DirectFieldReference implements FieldReference {
 
 		private final ExposedField field;
 
@@ -350,17 +375,16 @@ public final class ExposedFields implements Iterable<ExposedField> {
 		 * 
 		 * @param field must not be {@literal null}.
 		 */
-		public FieldReference(ExposedField field) {
+		public DirectFieldReference(ExposedField field) {
 
 			Assert.notNull(field, "ExposedField must not be null!");
 
 			this.field = field;
 		}
 
-		/**
-		 * Returns the raw, unqualified reference, i.e. the field reference without a {@literal $} prefix.
-		 * 
-		 * @return
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.aggregation.ExposedFields.FieldReference#getRaw()
 		 */
 		public String getRaw() {
 
@@ -368,11 +392,9 @@ public final class ExposedFields implements Iterable<ExposedField> {
 			return field.synthetic ? target : String.format("%s.%s", Fields.UNDERSCORE_ID, target);
 		}
 
-		/**
-		 * Returns the reference value for the given field reference. Will return 1 for a synthetic, unaliased field or the
-		 * raw rendering of the reference otherwise.
-		 * 
-		 * @return
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.aggregation.ExposedFields.FieldReference#getReferenceValue()
 		 */
 		public Object getReferenceValue() {
 			return field.synthetic && !field.isAliased() ? 1 : toString();
@@ -398,11 +420,11 @@ public final class ExposedFields implements Iterable<ExposedField> {
 				return true;
 			}
 
-			if (!(obj instanceof FieldReference)) {
+			if (!(obj instanceof DirectFieldReference)) {
 				return false;
 			}
 
-			FieldReference that = (FieldReference) obj;
+			DirectFieldReference that = (DirectFieldReference) obj;
 
 			return this.field.equals(that.field);
 		}
@@ -416,4 +438,78 @@ public final class ExposedFields implements Iterable<ExposedField> {
 			return field.hashCode();
 		}
 	}
+
+	/**
+	 * A {@link FieldReference} to a {@link Field} used within a nested {@link AggregationExpression}.
+	 *
+	 * @author Christoph Strobl
+	 * @since 1.10
+	 */
+	static class ExpressionFieldReference implements FieldReference {
+
+		private FieldReference delegate;
+
+		/**
+		 * Creates a new {@link FieldReference} for the given {@link ExposedField}.
+		 *
+		 * @param field must not be {@literal null}.
+		 */
+		public ExpressionFieldReference(FieldReference field) {
+			delegate = field;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.aggregation.ExposedFields.FieldReference#getRaw()
+		 */
+		@Override
+		public String getRaw() {
+			return delegate.getRaw();
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.aggregation.ExposedFields.FieldReference#getReferenceValue()
+		 */
+		@Override
+		public Object getReferenceValue() {
+			return delegate.getReferenceValue();
+		}
+
+		@Override
+		public String toString() {
+
+			String fieldRef = delegate.toString();
+
+			if (fieldRef.startsWith("$$")) {
+				return fieldRef;
+			}
+
+			if (fieldRef.startsWith("$")) {
+				return "$" + fieldRef;
+			}
+
+			return fieldRef;
+		}
+
+		@Override
+		public boolean equals(Object obj) {
+
+			if (this == obj) {
+				return true;
+			}
+
+			if (!(obj instanceof ExpressionFieldReference)) {
+				return false;
+			}
+
+			ExpressionFieldReference that = (ExpressionFieldReference) obj;
+			return ObjectUtils.nullSafeEquals(this.delegate, that.delegate);
+		}
+
+		@Override
+		public int hashCode() {
+			return delegate.hashCode();
+		}
+	}
 }

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

@@ -17,6 +17,7 @@ package org.springframework.data.mongodb.core.aggregation;
 
 import org.bson.Document;
 import org.springframework.data.mongodb.core.aggregation.ExposedFields.ExposedField;
+import org.springframework.data.mongodb.core.aggregation.ExposedFields.DirectFieldReference;
 import org.springframework.data.mongodb.core.aggregation.ExposedFields.FieldReference;
 import org.springframework.util.Assert;
 
@@ -112,10 +113,10 @@ class ExposedFieldsAggregationOperationContext implements AggregationOperationCo
 
 			if (field != null) {
 				// we return a FieldReference to the given field directly to make sure that we reference the proper alias here.
-				return new FieldReference(new ExposedField(field, exposedField.isSynthetic()));
+				return new DirectFieldReference(new ExposedField(field, exposedField.isSynthetic()));
 			}
 
-			return new FieldReference(exposedField);
+			return new DirectFieldReference(exposedField);
 		}
 
 		if (name.contains(".")) {
@@ -126,7 +127,7 @@ class ExposedFieldsAggregationOperationContext implements AggregationOperationCo
 			if (rootField != null) {
 
 				// We have to synthetic to true, in order to render the field-name as is.
-				return new FieldReference(new ExposedField(name, true));
+				return new DirectFieldReference(new ExposedField(name, true));
 			}
 		}
 		return null;

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

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2016. 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
+ *
+ *      http://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.aggregation;
+
+import org.bson.Document;
+import org.springframework.data.mongodb.core.aggregation.ExposedFields.FieldReference;
+import org.springframework.data.mongodb.core.aggregation.ExposedFields.ExpressionFieldReference;
+import org.springframework.util.Assert;
+
+import com.mongodb.DBObject;
+
+/**
+ * {@link AggregationOperationContext} that delegates {@link FieldReference} resolution and mapping to a parent one, but
+ * assures {@link FieldReference} get converted into {@link ExpressionFieldReference} using {@code $$} to ref an inner
+ * variable.
+ *
+ * @author Christoph Strobl
+ * @since 1.10
+ */
+class NestedDelegatingExpressionAggregationOperationContext implements AggregationOperationContext {
+
+	private final AggregationOperationContext delegate;
+
+	/**
+	 * Creates new {@link NestedDelegatingExpressionAggregationOperationContext}.
+	 *
+	 * @param referenceContext must not be {@literal null}.
+	 */
+	public NestedDelegatingExpressionAggregationOperationContext(AggregationOperationContext referenceContext) {
+
+		Assert.notNull(referenceContext, "Reference context must not be null!");
+		this.delegate = referenceContext;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperationContext#getMappedObject(org.bson.Document)
+	 */
+	@Override
+	public Document getMappedObject(Document document) {
+		return delegate.getMappedObject(document);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperationContext#getReference(org.springframework.data.mongodb.core.aggregation.Field)
+	 */
+	@Override
+	public FieldReference getReference(Field field) {
+		return new ExpressionFieldReference(delegate.getReference(field));
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperationContext#getReference(java.lang.String)
+	 */
+	@Override
+	public FieldReference getReference(String name) {
+		return new ExpressionFieldReference(delegate.getReference(name));
+	}
+}

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

@@ -639,6 +639,19 @@ public class ProjectionOperation implements FieldsExposingAggregationOperation {
 			return project("slice", offset, count);
 		}
 
+		/**
+		 * Generates a {@code $filter} expression that returns a subset of the array held by the given field.
+		 *
+		 * @param as The variable name for the element in the input array. Must not be {@literal null}.
+		 * @param condition The {@link AggregationExpression} that determines whether to include the element in the
+		 *          resulting array. Must not be {@literal null}.
+		 * @return never {@literal null}.
+		 * @since 1.10
+		 */
+		public ProjectionOperationBuilder filter(String as, AggregationExpression condition) {
+			return this.operation.and(AggregationExpressions.Filter.filter(name).as(as).by(condition));
+		}
+
 		/* 
 		 * (non-Javadoc)
 		 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#toDocument(org.springframework.data.mongodb.core.aggregation.AggregationOperationContext)

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

@@ -23,6 +23,7 @@ import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.data.mapping.context.PersistentPropertyPath;
 import org.springframework.data.mongodb.core.aggregation.ExposedFields.ExposedField;
 import org.springframework.data.mongodb.core.aggregation.ExposedFields.FieldReference;
+import org.springframework.data.mongodb.core.aggregation.ExposedFields.DirectFieldReference;
 import org.springframework.data.mongodb.core.convert.QueryMapper;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentEntity;
 import org.springframework.data.mongodb.core.mapping.MongoPersistentProperty;
@@ -98,6 +99,6 @@ public class TypeBasedAggregationOperationContext implements AggregationOperatio
 		Field mappedField = field(propertyPath.getLeafProperty().getName(),
 				propertyPath.toDotPath(MongoPersistentProperty.PropertyToFieldNameConverter.INSTANCE));
 
-		return new FieldReference(new ExposedField(mappedField, true));
+		return new DirectFieldReference(new ExposedField(mappedField, true));
 	}
 }

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

@@ -29,6 +29,7 @@ import java.text.ParseException;
 import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Date;
 import java.util.List;
 import java.util.Scanner;
@@ -68,6 +69,8 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import com.mongodb.MongoException;
 import com.mongodb.client.MongoCollection;
 
+import lombok.Builder;
+
 /**
  * Tests for {@link MongoTemplate#aggregate(String, AggregationPipeline, Class)}.
  * 
@@ -1496,6 +1499,42 @@ public class AggregationTests {
 		}
 	}
 
+	/**
+	 * @see DATAMONGO-1491
+	 */
+	@Test
+	public void filterShouldBeAppliedCorrectly() {
+
+		assumeTrue(mongoVersion.isGreaterThanOrEqualTo(THREE_DOT_TWO));
+
+		Item item43 = Item.builder().itemId("43").quantity(2).price(2L).build();
+		Item item2 = Item.builder().itemId("2").quantity(1).price(240L).build();
+		Sales sales1 = Sales.builder().id("0")
+				.items(Arrays.asList( //
+						item43, item2)) //
+				.build();
+
+		Item item23 = Item.builder().itemId("23").quantity(3).price(110L).build();
+		Item item103 = Item.builder().itemId("103").quantity(4).price(5L).build();
+		Item item38 = Item.builder().itemId("38").quantity(1).price(300L).build();
+		Sales sales2 = Sales.builder().id("1").items(Arrays.asList( //
+				item23, item103, item38)).build();
+
+		Item item4 = Item.builder().itemId("4").quantity(1).price(23L).build();
+		Sales sales3 = Sales.builder().id("2").items(Arrays.asList( //
+				item4)).build();
+
+		mongoTemplate.insert(Arrays.asList(sales1, sales2, sales3), Sales.class);
+
+		TypedAggregation<Sales> agg = newAggregation(Sales.class, project().and("items")
+				.filter("item", AggregationFunctionExpressions.GTE.of(field("item.price"), 100)).as("items"));
+
+		assertThat(mongoTemplate.aggregate(agg, Sales.class).getMappedResults(),
+				contains(Sales.builder().id("0").items(Collections.singletonList(item2)).build(),
+						Sales.builder().id("1").items(Arrays.asList(item23, item38)).build(),
+						Sales.builder().id("2").items(Collections.<Item> emptyList()).build()));
+	}
+
 	private void createUsersWithReferencedPersons() {
 
 		mongoTemplate.dropCollection(User.class);
@@ -1736,4 +1775,22 @@ public class AggregationTests {
 			this.qty = qty;
 		}
 	}
+
+	@lombok.Data
+	@Builder
+	static class Sales {
+
+		@Id String id;
+		List<Item> items;
+	}
+
+	@lombok.Data
+	@Builder
+	static class Item {
+
+		@org.springframework.data.mongodb.core.mapping.Field("item_id") //
+		String itemId;
+		Integer quantity;
+		Long price;
+	}
 }

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

@@ -0,0 +1,139 @@
+/*
+ * Copyright 2016. 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
+ *
+ *      http://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.aggregation;
+
+import static org.hamcrest.core.Is.*;
+import static org.junit.Assert.*;
+import static org.springframework.data.mongodb.core.aggregation.AggregationExpressions.Filter.*;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.bson.Document;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+import org.springframework.data.mongodb.MongoDbFactory;
+import org.springframework.data.mongodb.core.DocumentTestUtils;
+import org.springframework.data.mongodb.core.convert.DefaultDbRefResolver;
+import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
+import org.springframework.data.mongodb.core.convert.QueryMapper;
+import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
+
+/**
+ * @author Christoph Strobl
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class FilterExpressionUnitTests {
+
+	@Mock MongoDbFactory mongoDbFactory;
+
+	private AggregationOperationContext aggregationContext;
+	private MongoMappingContext mappingContext;
+
+	@Before
+	public void setUp() {
+
+		mappingContext = new MongoMappingContext();
+		aggregationContext = new TypeBasedAggregationOperationContext(Sales.class, mappingContext,
+				new QueryMapper(new MappingMongoConverter(new DefaultDbRefResolver(mongoDbFactory), mappingContext)));
+	}
+
+	/**
+	 * @see DATAMONGO-1491
+	 */
+	@Test
+	public void shouldConstructFilterExpressionCorrectly() {
+
+		TypedAggregation<Sales> agg = Aggregation.newAggregation(Sales.class,
+				Aggregation.project()
+						.and(filter("items").as("item").by(AggregationFunctionExpressions.GTE.of(Fields.field("item.price"), 100)))
+						.as("items"));
+
+		Document dbo = agg.toDocument("sales", aggregationContext);
+
+		List<Object> pipeline = DocumentTestUtils.getAsDBList(dbo, "pipeline");
+		Document $project = DocumentTestUtils.getAsDocument((Document) pipeline.get(0), "$project");
+		Document items = DocumentTestUtils.getAsDocument($project, "items");
+		Document $filter = DocumentTestUtils.getAsDocument(items, "$filter");
+
+		Document expected = Document.parse("{" + //
+				"input: \"$items\"," + //
+				"as: \"item\"," + //
+				"cond: { $gte: [ \"$$item.price\", 100 ] }" + //
+				"}");
+
+		assertThat($filter, is(new Document(expected)));
+	}
+
+	/**
+	 * @see DATAMONGO-1491
+	 */
+	@Test
+	public void shouldConstructFilterExpressionCorrectlyWhenUsingFilterOnProjectionBuilder() {
+
+		TypedAggregation<Sales> agg = Aggregation.newAggregation(Sales.class, Aggregation.project().and("items")
+				.filter("item", AggregationFunctionExpressions.GTE.of(Fields.field("item.price"), 100)).as("items"));
+
+		Document dbo = agg.toDocument("sales", aggregationContext);
+
+		List<Object> pipeline = DocumentTestUtils.getAsDBList(dbo, "pipeline");
+		Document $project = DocumentTestUtils.getAsDocument((Document) pipeline.get(0), "$project");
+		Document items = DocumentTestUtils.getAsDocument($project, "items");
+		Document $filter = DocumentTestUtils.getAsDocument(items, "$filter");
+
+		Document expected = Document.parse("{" + //
+				"input: \"$items\"," + //
+				"as: \"item\"," + //
+				"cond: { $gte: [ \"$$item.price\", 100 ] }" + //
+				"}");
+
+		assertThat($filter, is(expected));
+	}
+
+	/**
+	 * @see DATAMONGO-1491
+	 */
+	@Test
+	public void shouldConstructFilterExpressionCorrectlyWhenInputMapToArray() {
+
+		TypedAggregation<Sales> agg = Aggregation.newAggregation(Sales.class,
+				Aggregation.project().and(filter(Arrays.<Object> asList(1, "a", 2, null, 3.1D, 4, "5")).as("num")
+						.by(AggregationFunctionExpressions.GTE.of(Fields.field("num"), 3))).as("items"));
+
+		Document dbo = agg.toDocument("sales", aggregationContext);
+
+		List<Object> pipeline = DocumentTestUtils.getAsDBList(dbo, "pipeline");
+		Document $project = DocumentTestUtils.getAsDocument((Document) pipeline.get(0), "$project");
+		Document items = DocumentTestUtils.getAsDocument($project, "items");
+		Document $filter = DocumentTestUtils.getAsDocument(items, "$filter");
+
+		Document expected = Document.parse("{" + //
+				"input: [ 1, \"a\", 2, null, 3.1, 4, \"5\" ]," + //
+				"as: \"num\"," + //
+				"cond: { $gte: [ \"$$num\", 3 ] }" + //
+				"}");
+
+		assertThat($filter, is(expected));
+	}
+
+	static class Sales {
+
+		List<Object> items;
+	}
+}

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

@@ -39,6 +39,7 @@ import org.springframework.data.domain.Sort.Direction;
 import org.springframework.data.mapping.model.MappingException;
 import org.springframework.data.mongodb.core.aggregation.ExposedFields.ExposedField;
 import org.springframework.data.mongodb.core.aggregation.ExposedFields.FieldReference;
+import org.springframework.data.mongodb.core.aggregation.ExposedFields.DirectFieldReference;
 import org.springframework.data.mongodb.core.convert.CustomConversions;
 import org.springframework.data.mongodb.core.convert.DbRefResolver;
 import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
@@ -102,7 +103,7 @@ public class TypeBasedAggregationOperationContextUnitTests {
 	public void aliasesIdFieldCorrectly() {
 
 		AggregationOperationContext context = getContext(Foo.class);
-		assertThat(context.getReference("id"), is(new FieldReference(new ExposedField(field("id", "_id"), true))));
+		assertThat(context.getReference("id"), is((FieldReference) new DirectFieldReference(new ExposedField(field("id", "_id"), true))));
 	}
 
 	/**

src/main/asciidoc/reference/mongodb.adoc

@@ -1688,7 +1688,7 @@ At the time of this writing we provide support for the following Aggregation Ope
 | eq (*via: is), gt, gte, lt, lte, ne
 
 | Array Aggregation Operators
-| size, slice
+| size, slice, filter
 
 | Conditional Aggregation Operators
 | cond, ifNull