Add support for $setWindowFields aggregation stage.
Add a SetWindowFieldsOperation to the aggregation framework.
The builder API allows fluent declaration of the aggregation stage as shown in the sample below.
SetWindowFieldsOperation.builder()
.partitionByField("state")
.sortBy(Sort.by(Direction.ASC, "date"))
.output(AccumulatorOperators.valueOf("qty").sum())
.within(Windows.documents().fromUnbounded().toCurrent().build())
.as("cumulativeQuantityForState")
.build();
Closes #3711
Original pull request: #3739.
This commit is contained in:
Christoph Strobl
committed by
Mark Paluch
Mark Paluch
parent
255491c446
commit
23254c10dc
@@ -0,0 +1,783 @@
|
||||
/*
|
||||
* 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.aggregation;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
import org.bson.Document;
|
||||
import org.springframework.data.domain.Sort;
|
||||
import org.springframework.lang.Nullable;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* @author Christoph Strobl
|
||||
* @since 3.3
|
||||
* @see <a href=
|
||||
* "https://docs.mongodb.com/manual/reference/operator/aggregation/setWindowFields/">https://docs.mongodb.com/manual/reference/operator/aggregation/setWindowFields/</a>
|
||||
*/
|
||||
public class SetWindowFieldsOperation
|
||||
implements AggregationOperation, FieldsExposingAggregationOperation.InheritsFieldsAggregationOperation {
|
||||
|
||||
@Nullable //
|
||||
private Object partitionBy;
|
||||
|
||||
@Nullable //
|
||||
private AggregationOperation sortBy;
|
||||
|
||||
private WindowOutput output;
|
||||
|
||||
/**
|
||||
* Create a new {@link SetWindowFieldsOperation} with given args.
|
||||
*
|
||||
* @param partitionBy The field or {@link AggregationExpression} to group by.
|
||||
* @param sortBy the {@link SortOperation operation} to sort the documents by in the partition.
|
||||
* @param output the {@link WindowOutput} containing the fields to add and the rules to calculate their respective
|
||||
* values.
|
||||
*/
|
||||
public SetWindowFieldsOperation(@Nullable Object partitionBy, @Nullable AggregationOperation sortBy,
|
||||
WindowOutput output) {
|
||||
|
||||
this.partitionBy = partitionBy;
|
||||
this.sortBy = sortBy;
|
||||
this.output = output;
|
||||
}
|
||||
|
||||
/**
|
||||
* Obtain a {@link SetWindowFieldsOperationBuilder builder} to create a {@link SetWindowFieldsOperation}.
|
||||
*
|
||||
* @return new instance of {@link SetWindowFieldsOperationBuilder}.
|
||||
*/
|
||||
public static SetWindowFieldsOperationBuilder builder() {
|
||||
return new SetWindowFieldsOperationBuilder();
|
||||
}
|
||||
|
||||
@Override
|
||||
public ExposedFields getFields() {
|
||||
return ExposedFields.nonSynthetic(Fields.from(output.fields.toArray(new Field[0])));
|
||||
}
|
||||
|
||||
@Override
|
||||
public Document toDocument(AggregationOperationContext context) {
|
||||
|
||||
Document $setWindowFields = new Document();
|
||||
if (partitionBy != null) {
|
||||
if (partitionBy instanceof AggregationExpression) {
|
||||
$setWindowFields.append("partitionBy", ((AggregationExpression) partitionBy).toDocument(context));
|
||||
} else if (partitionBy instanceof Field) {
|
||||
$setWindowFields.append("partitionBy", context.getReference((Field) partitionBy).toString());
|
||||
} else {
|
||||
$setWindowFields.append("partitionBy", partitionBy);
|
||||
}
|
||||
}
|
||||
|
||||
if (sortBy != null) {
|
||||
$setWindowFields.append("sortBy", sortBy.toDocument(context).get(sortBy.getOperator()));
|
||||
}
|
||||
|
||||
Document output = new Document();
|
||||
for (ComputedField field : this.output.fields) {
|
||||
|
||||
Document fieldOperation = field.getWindowOperator().toDocument(context);
|
||||
if (field.window != null) {
|
||||
fieldOperation.put("window", field.window.toDocument(context));
|
||||
}
|
||||
output.append(field.getName(), fieldOperation);
|
||||
}
|
||||
$setWindowFields.append("output", output);
|
||||
|
||||
return new Document(getOperator(), $setWindowFields);
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#getOperator()
|
||||
*/
|
||||
@Override
|
||||
public String getOperator() {
|
||||
return "$setWindowFields";
|
||||
}
|
||||
|
||||
/**
|
||||
* {@link WindowOutput} defines output of {@literal $setWindowFields} stage by defining the {@link ComputedField
|
||||
* field(s)} to append to the documents in the output.
|
||||
*/
|
||||
public static class WindowOutput {
|
||||
|
||||
private List<ComputedField> fields;
|
||||
|
||||
/**
|
||||
* Create a new output containing the single given {@link ComputedField field}.
|
||||
*
|
||||
* @param outputField must not be {@literal null}.
|
||||
*/
|
||||
public WindowOutput(ComputedField outputField) {
|
||||
|
||||
Assert.notNull(outputField, "OutputField must not be null!");
|
||||
this.fields = new ArrayList<>();
|
||||
this.fields.add(outputField);
|
||||
}
|
||||
|
||||
/**
|
||||
* Append the given {@link ComputedField field} to the outptut.
|
||||
*
|
||||
* @param field must not be {@literal null}.
|
||||
* @return this.
|
||||
*/
|
||||
public WindowOutput append(ComputedField field) {
|
||||
|
||||
Assert.notNull(field, "Field must not be null!");
|
||||
fields.add(field);
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Append the given {@link AggregationExpression} as a {@link ComputedField field} in a fluent way.
|
||||
*
|
||||
* @param expression must not be {@literal null}.
|
||||
* @return new instance of {@link ComputedFieldAppender}.
|
||||
* @see #append(ComputedField)
|
||||
*/
|
||||
public ComputedFieldAppender append(AggregationExpression expression) {
|
||||
|
||||
return new ComputedFieldAppender() {
|
||||
|
||||
@Nullable private Window window;
|
||||
|
||||
@Override
|
||||
public WindowOutput as(String fieldname) {
|
||||
|
||||
return WindowOutput.this.append(new ComputedField(fieldname, expression, window));
|
||||
}
|
||||
|
||||
@Override
|
||||
public ComputedFieldAppender within(Window window) {
|
||||
this.window = window;
|
||||
return this;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Tiny little helper to allow fluent API usage for {@link #append(ComputedField)}.
|
||||
*/
|
||||
interface ComputedFieldAppender {
|
||||
|
||||
/**
|
||||
* Specify the target field name.
|
||||
*
|
||||
* @param fieldname the name of field to add to the target document.
|
||||
* @return the {@link WindowOutput} that started the append operation.
|
||||
*/
|
||||
WindowOutput as(String fieldname);
|
||||
|
||||
/**
|
||||
* Specify the window boundaries.
|
||||
*
|
||||
* @param window must not be {@literal null}.
|
||||
* @return this.
|
||||
*/
|
||||
ComputedFieldAppender within(Window window);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A {@link Field} that the result of a computation done via an {@link AggregationExpression}.
|
||||
*
|
||||
* @author Christoph Strobl
|
||||
*/
|
||||
public static class ComputedField implements Field {
|
||||
|
||||
private String name;
|
||||
private AggregationExpression windowOperator;
|
||||
|
||||
@Nullable //
|
||||
private Window window;
|
||||
|
||||
/**
|
||||
* Create a new {@link ComputedField}.
|
||||
*
|
||||
* @param name the target field name.
|
||||
* @param windowOperator the expression to calculate the field value.
|
||||
*/
|
||||
public ComputedField(String name, AggregationExpression windowOperator) {
|
||||
this(name, windowOperator, null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new {@link ComputedField}.
|
||||
*
|
||||
* @param name the target field name.
|
||||
* @param windowOperator the expression to calculate the field value.
|
||||
* @param window the boundaries to operate within. Can be {@literal null}.
|
||||
*/
|
||||
public ComputedField(String name, AggregationExpression windowOperator, @Nullable Window window) {
|
||||
|
||||
this.name = name;
|
||||
this.windowOperator = windowOperator;
|
||||
this.window = window;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getTarget() {
|
||||
return getName();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isAliased() {
|
||||
return false;
|
||||
}
|
||||
|
||||
public AggregationExpression getWindowOperator() {
|
||||
return windowOperator;
|
||||
}
|
||||
|
||||
public Window getWindow() {
|
||||
return window;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Quick access to {@link DocumentWindow documents} and {@literal RangeWindow range} {@link Window windows}.
|
||||
*
|
||||
* @author Christoph Strobl
|
||||
*/
|
||||
public interface Windows {
|
||||
|
||||
/**
|
||||
* Create a document window relative to the position of the current document.
|
||||
*
|
||||
* @param lower an integer for a position relative to the current document, {@literal current} or
|
||||
* {@literal unbounded}.
|
||||
* @param upper an integer for a position relative to the current document, {@literal current} or
|
||||
* {@literal unbounded}.
|
||||
* @return new instance of {@link DocumentWindow}.
|
||||
*/
|
||||
static DocumentWindow documents(Object lower, Object upper) {
|
||||
return new DocumentWindow(lower, upper);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a range window defined based on sort expression.
|
||||
*
|
||||
* @param lower a numeric value to add the sort by field value of the current document, {@literal current} or
|
||||
* {@literal unbounded}.
|
||||
* @param upper a numeric value to add the sort by field value of the current document, {@literal current} or
|
||||
* {@literal unbounded}.
|
||||
* @return new instance of {@link RangeWindow}.
|
||||
*/
|
||||
static RangeWindow range(Object lower, Object upper, @Nullable WindowUnit unit) {
|
||||
return new RangeWindow(lower, upper, unit);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a range window based on the {@link Sort sort value} of the current document via a fluent API.
|
||||
*
|
||||
* @return new instance of {@link RangeWindowBuilder}.
|
||||
*/
|
||||
static RangeWindowBuilder range() {
|
||||
return new RangeWindowBuilder();
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a document window relative to the position of the current document via a fluent API.
|
||||
*
|
||||
* @return new instance of {@link DocumentWindowBuilder}.
|
||||
*/
|
||||
static DocumentWindowBuilder documents() {
|
||||
return new DocumentWindowBuilder();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A {@link Window} to be used for {@link ComputedField#getWindow() ComputedField}.
|
||||
*/
|
||||
public interface Window {
|
||||
|
||||
/**
|
||||
* The upper (inclusive) boundary.
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
Object getUpper();
|
||||
|
||||
/**
|
||||
* The lower (inclusive) boundary.
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
Object getLower();
|
||||
|
||||
/**
|
||||
* Obtain the document representation of the window in a default {@link AggregationOperationContext context}.
|
||||
*
|
||||
* @return never {@literal null}.
|
||||
*/
|
||||
default Document toDocument() {
|
||||
return toDocument(Aggregation.DEFAULT_CONTEXT);
|
||||
}
|
||||
|
||||
/**
|
||||
* Obtain the document representation of the window in the given {@link AggregationOperationContext context}.
|
||||
*
|
||||
* @return never {@literal null}.
|
||||
*/
|
||||
Document toDocument(AggregationOperationContext ctx);
|
||||
}
|
||||
|
||||
/**
|
||||
* Builder API for a {@link RangeWindow}.
|
||||
*
|
||||
* @author Christoph Strobl
|
||||
*/
|
||||
public static class RangeWindowBuilder {
|
||||
|
||||
@Nullable //
|
||||
private Object upper;
|
||||
|
||||
@Nullable //
|
||||
private Object lower;
|
||||
|
||||
@Nullable //
|
||||
private WindowUnit unit;
|
||||
|
||||
/**
|
||||
* The upper (inclusive) range limit based on the sortBy field.
|
||||
*
|
||||
* @param upper eg. {@literal current} or {@literal unbounded}.
|
||||
* @return this.
|
||||
*/
|
||||
public RangeWindowBuilder to(String upper) {
|
||||
|
||||
this.upper = upper;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* The lower (inclusive) range limit based on the sortBy field.
|
||||
*
|
||||
* @param lower eg. {@literal current} or {@literal unbounded}.
|
||||
* @return this.
|
||||
*/
|
||||
public RangeWindowBuilder from(String lower) {
|
||||
|
||||
this.lower = lower;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* The upper (inclusive) range limit value to add to the value based on the sortBy field.
|
||||
*
|
||||
* @param upper
|
||||
* @return this.
|
||||
*/
|
||||
public RangeWindowBuilder to(Number upper) {
|
||||
|
||||
this.upper = upper;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* The lower (inclusive) range limit value to add to the value based on the sortBy field.
|
||||
*
|
||||
* @param lower
|
||||
* @return this.
|
||||
*/
|
||||
public RangeWindowBuilder from(Number lower) {
|
||||
|
||||
this.lower = lower;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Use {@literal current} as {@link #from(String) lower} limit.
|
||||
*
|
||||
* @return this.
|
||||
*/
|
||||
public RangeWindowBuilder fromCurrent() {
|
||||
return from("current");
|
||||
}
|
||||
|
||||
/**
|
||||
* Use {@literal unbounded} as {@link #from(String) lower} limit.
|
||||
*
|
||||
* @return this.
|
||||
*/
|
||||
public RangeWindowBuilder fromUnbounded() {
|
||||
return from("unbounded");
|
||||
}
|
||||
|
||||
/**
|
||||
* Use {@literal current} as {@link #to(String) upper} limit.
|
||||
*
|
||||
* @return this.
|
||||
*/
|
||||
public RangeWindowBuilder toCurrent() {
|
||||
return to("current");
|
||||
}
|
||||
|
||||
/**
|
||||
* Use {@literal unbounded} as {@link #to(String) upper} limit.
|
||||
*
|
||||
* @return this.
|
||||
*/
|
||||
public RangeWindowBuilder toUnbounded() {
|
||||
return to("unbounded");
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the {@link WindowUnit unit} or measure for the given {@link Window}.
|
||||
*
|
||||
* @param windowUnit must not be {@literal null}. Can be on of {@link Windows}.
|
||||
* @return this.
|
||||
*/
|
||||
public RangeWindowBuilder unit(WindowUnit windowUnit) {
|
||||
|
||||
this.unit = unit;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the {@link RangeWindow}.
|
||||
*
|
||||
* @return new instance of {@link RangeWindow}.
|
||||
*/
|
||||
public RangeWindow build() {
|
||||
return new RangeWindow(lower, upper, unit);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Builder API for a {@link RangeWindow}.
|
||||
*
|
||||
* @author Christoph Strobl
|
||||
*/
|
||||
public static class DocumentWindowBuilder {
|
||||
|
||||
@Nullable //
|
||||
private Object upper;
|
||||
|
||||
@Nullable //
|
||||
private Object lower;
|
||||
|
||||
public DocumentWindowBuilder from(Number lower) {
|
||||
|
||||
this.lower = lower;
|
||||
return this;
|
||||
}
|
||||
|
||||
public DocumentWindowBuilder fromCurrent() {
|
||||
return from("current");
|
||||
}
|
||||
|
||||
public DocumentWindowBuilder fromUnbounded() {
|
||||
return from("unbounded");
|
||||
}
|
||||
|
||||
public DocumentWindowBuilder to(String upper) {
|
||||
|
||||
this.upper = upper;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* The lower (inclusive) range limit based on current document.
|
||||
*
|
||||
* @param lower eg. {@literal current} or {@literal unbounded}.
|
||||
* @return this.
|
||||
*/
|
||||
public DocumentWindowBuilder from(String lower) {
|
||||
|
||||
this.lower = lower;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* The upper (inclusive) range limit based on current document.
|
||||
*
|
||||
* @param upper eg. {@literal current} or {@literal unbounded}.
|
||||
* @return this.
|
||||
*/
|
||||
public DocumentWindowBuilder to(Number upper) {
|
||||
|
||||
this.upper = upper;
|
||||
return this;
|
||||
}
|
||||
|
||||
public DocumentWindowBuilder toCurrent() {
|
||||
return to("current");
|
||||
}
|
||||
|
||||
public DocumentWindowBuilder toUnbounded() {
|
||||
return to("unbounded");
|
||||
}
|
||||
|
||||
public DocumentWindow build() {
|
||||
return new DocumentWindow(lower, upper);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Common base class for {@link Window} implementation.
|
||||
*
|
||||
* @author Christoph Strobl
|
||||
*/
|
||||
abstract static class WindowImp implements Window {
|
||||
|
||||
private final Object upper;
|
||||
private final Object lower;
|
||||
|
||||
protected WindowImp(Object lower, Object upper) {
|
||||
this.upper = upper;
|
||||
this.lower = lower;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object getUpper() {
|
||||
return upper;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object getLower() {
|
||||
return lower;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* {@link Window} implementation based on the current document.
|
||||
*
|
||||
* @author Christoph Strobl
|
||||
*/
|
||||
public static class DocumentWindow extends WindowImp {
|
||||
|
||||
DocumentWindow(Object lower, Object upper) {
|
||||
super(lower, upper);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Document toDocument(AggregationOperationContext ctx) {
|
||||
return new Document("documents", Arrays.asList(getLower(), getUpper()));
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* {@link Window} implementation based on the sort fields.
|
||||
*
|
||||
* @author Christoph Strobl
|
||||
*/
|
||||
public static class RangeWindow extends WindowImp {
|
||||
|
||||
@Nullable //
|
||||
private WindowUnit unit;
|
||||
|
||||
protected RangeWindow(Object lower, Object upper, WindowUnit unit) {
|
||||
|
||||
super(lower, upper);
|
||||
this.unit = unit;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Document toDocument(AggregationOperationContext ctx) {
|
||||
|
||||
Document range = new Document("range", new Object[] { getLower(), getUpper() });
|
||||
if (unit != null && !WindowUnits.DEFAULT.equals(unit)) {
|
||||
range.append("unit", unit.name().toLowerCase());
|
||||
}
|
||||
return range;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The actual time unit to apply to a {@link Window}.
|
||||
*/
|
||||
public interface WindowUnit {
|
||||
String name();
|
||||
}
|
||||
|
||||
/**
|
||||
* Quick access to available {@link WindowUnit units}.
|
||||
*/
|
||||
public enum WindowUnits implements WindowUnit {
|
||||
DEFAULT, YEAR, QUARTER, MONTH, WEEK, DAY, HOUR, MINUTE, SECOND, MILLISECOND
|
||||
}
|
||||
|
||||
/**
|
||||
* A fluent builder to create a {@link SetWindowFieldsOperation}.
|
||||
*
|
||||
* @author Christoph Strobl
|
||||
*/
|
||||
public static class SetWindowFieldsOperationBuilder {
|
||||
|
||||
private Object partitionBy;
|
||||
private SortOperation sortOperation;
|
||||
private WindowOutput output;
|
||||
|
||||
/**
|
||||
* Specify the field to group by.
|
||||
*
|
||||
* @param fieldName must not be {@literal null}.
|
||||
* @return this.
|
||||
*/
|
||||
public SetWindowFieldsOperationBuilder partitionByField(String fieldName) {
|
||||
return partitionBy(Fields.field("$" + fieldName, fieldName));
|
||||
}
|
||||
|
||||
/**
|
||||
* Specify the {@link AggregationExpression expression} to group by.
|
||||
*
|
||||
* @param expression must not be {@literal null}.
|
||||
* @return this.
|
||||
*/
|
||||
public SetWindowFieldsOperationBuilder partitionByExpression(AggregationExpression expression) {
|
||||
return partitionBy(expression);
|
||||
}
|
||||
|
||||
/**
|
||||
* Sort {@link Sort.Direction#ASC ascending} by the given fields.
|
||||
*
|
||||
* @param fields must not be {@literal null}.
|
||||
* @return this.
|
||||
*/
|
||||
public SetWindowFieldsOperationBuilder sortBy(String... fields) {
|
||||
return sortBy(Sort.by(fields));
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the sort order.
|
||||
*
|
||||
* @param sort must not be {@literal null}.
|
||||
* @return this.
|
||||
*/
|
||||
public SetWindowFieldsOperationBuilder sortBy(Sort sort) {
|
||||
return sortBy(new SortOperation(sort));
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the {@link SortOperation} to use.
|
||||
*
|
||||
* @param sort must not be {@literal null}.
|
||||
* @return this.
|
||||
*/
|
||||
public SetWindowFieldsOperationBuilder sortBy(SortOperation sort) {
|
||||
|
||||
this.sortOperation = sort;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Define the actual output computation.
|
||||
*
|
||||
* @param output must not be {@literal null}.
|
||||
* @return this.
|
||||
*/
|
||||
public SetWindowFieldsOperationBuilder output(WindowOutput output) {
|
||||
|
||||
this.output = output;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a field capturing the result of the given {@link AggregationExpression expression} to the output.
|
||||
*
|
||||
* @param expression must not be {@literal null}.
|
||||
* @return new instance of {@link WindowChoice}.
|
||||
*/
|
||||
public WindowChoice output(AggregationExpression expression) {
|
||||
|
||||
return new WindowChoice() {
|
||||
|
||||
@Nullable private Window window;
|
||||
|
||||
@Override
|
||||
public As within(Window window) {
|
||||
|
||||
this.window = window;
|
||||
return this;
|
||||
}
|
||||
|
||||
@Override
|
||||
public SetWindowFieldsOperationBuilder as(String targetFieldName) {
|
||||
|
||||
ComputedField computedField = new ComputedField(targetFieldName, expression, window);
|
||||
|
||||
if (SetWindowFieldsOperationBuilder.this.output == null) {
|
||||
SetWindowFieldsOperationBuilder.this.output = new WindowOutput(computedField);
|
||||
} else {
|
||||
SetWindowFieldsOperationBuilder.this.output.append(computedField);
|
||||
}
|
||||
|
||||
return SetWindowFieldsOperationBuilder.this;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Interface to capture field name used to capture the computation result.
|
||||
*/
|
||||
public interface As {
|
||||
|
||||
/**
|
||||
* Define the target name field name to hold the computation result.
|
||||
*
|
||||
* @param targetFieldName must not be {@literal null}.
|
||||
* @return the starting point {@link SetWindowFieldsOperationBuilder builder} instance.
|
||||
*/
|
||||
SetWindowFieldsOperationBuilder as(String targetFieldName);
|
||||
}
|
||||
|
||||
/**
|
||||
* Interface to capture an optional {@link Window} applicable to the field computation.
|
||||
*/
|
||||
public interface WindowChoice extends As {
|
||||
|
||||
/**
|
||||
* Specify calculation boundaries.
|
||||
*
|
||||
* @param window must not be {@literal null}.
|
||||
* @return never {@literal null}.
|
||||
*/
|
||||
As within(Window window);
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Partition by a value that transaltes to a valid mongodb expression.
|
||||
*
|
||||
* @param value must not be {@literal null}.
|
||||
* @return this.
|
||||
*/
|
||||
public SetWindowFieldsOperationBuilder partitionBy(Object value) {
|
||||
|
||||
partitionBy = value;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* Obtain a new instance of {@link SetWindowFieldsOperation} with previously set arguments.
|
||||
*
|
||||
* @return new instance of {@link SetWindowFieldsOperation}.
|
||||
*/
|
||||
public SetWindowFieldsOperation build() {
|
||||
return new SetWindowFieldsOperation(partitionBy, sortOperation, output);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,132 @@
|
||||
/*
|
||||
* 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.aggregation;
|
||||
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.Date;
|
||||
import java.util.List;
|
||||
|
||||
import org.bson.Document;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.extension.ExtendWith;
|
||||
import org.springframework.data.annotation.Id;
|
||||
import org.springframework.data.domain.Sort;
|
||||
import org.springframework.data.domain.Sort.Direction;
|
||||
import org.springframework.data.mongodb.core.aggregation.DateOperators.Year;
|
||||
import org.springframework.data.mongodb.core.aggregation.SetWindowFieldsOperation.Windows;
|
||||
import org.springframework.data.mongodb.core.mapping.Field;
|
||||
import org.springframework.data.mongodb.test.util.EnableIfMongoServerVersion;
|
||||
import org.springframework.data.mongodb.test.util.MongoTemplateExtension;
|
||||
import org.springframework.data.mongodb.test.util.MongoTestTemplate;
|
||||
import org.springframework.data.mongodb.test.util.Template;
|
||||
|
||||
/**
|
||||
* @author Christoph Strobl
|
||||
*/
|
||||
@ExtendWith(MongoTemplateExtension.class)
|
||||
@EnableIfMongoServerVersion(isGreaterThanEqual = "5.0")
|
||||
class SetWindowFieldsOperationTests {
|
||||
|
||||
@Template //
|
||||
private static MongoTestTemplate mongoTemplate;
|
||||
|
||||
@AfterEach
|
||||
void afterEach() {
|
||||
mongoTemplate.flush(CakeSale.class);
|
||||
}
|
||||
|
||||
@Test // GH-3711
|
||||
void executesSetWindowFieldsOperationCorrectly() {
|
||||
|
||||
initCakeSales();
|
||||
|
||||
SetWindowFieldsOperation setWindowFieldsOperation = SetWindowFieldsOperation.builder() //
|
||||
.partitionByField("state") // resolves to field ref "$state"
|
||||
.sortBy(Sort.by(Direction.ASC, "date")) // resolves to "orderDate"
|
||||
.output(AccumulatorOperators.valueOf("qty").sum()) // resolves to "$quantity"
|
||||
.within(Windows.documents().fromUnbounded().toCurrent().build()) //
|
||||
.as("cumulativeQuantityForState") //
|
||||
.build(); //
|
||||
|
||||
AggregationResults<Document> results = mongoTemplate.aggregateAndReturn(Document.class)
|
||||
.by(Aggregation.newAggregation(CakeSale.class, setWindowFieldsOperation)).all();
|
||||
|
||||
assertThat(results.getMappedResults()).map(it -> it.get("cumulativeQuantityForState")).contains(162, 282, 427, 134,
|
||||
238, 378);
|
||||
}
|
||||
|
||||
@Test // GH-3711
|
||||
void executesSetWindowFieldsOperationWithPartitionExpressionCorrectly() {
|
||||
|
||||
initCakeSales();
|
||||
|
||||
SetWindowFieldsOperation setWindowFieldsOperation = SetWindowFieldsOperation.builder() //
|
||||
.partitionByExpression(Year.yearOf("date")) // resolves to $year: "$orderDate"
|
||||
.sortBy(Sort.by(Direction.ASC, "date")) // resolves to "orderDate"
|
||||
.output(AccumulatorOperators.valueOf("qty").sum()) // resolves to "$quantity"
|
||||
.within(Windows.documents().fromUnbounded().toCurrent().build()) //
|
||||
.as("cumulativeQuantityForState") //
|
||||
.build(); //
|
||||
|
||||
AggregationResults<Document> results = mongoTemplate.aggregateAndReturn(Document.class)
|
||||
.by(Aggregation.newAggregation(CakeSale.class, setWindowFieldsOperation)).all();
|
||||
|
||||
assertThat(results.getMappedResults()).map(it -> it.get("cumulativeQuantityForState")).contains(134, 296, 104, 224,
|
||||
145, 285);
|
||||
}
|
||||
|
||||
void initCakeSales() {
|
||||
|
||||
mongoTemplate.execute(CakeSale.class, collection -> {
|
||||
|
||||
List<Document> source = Arrays.asList(Document.parse(
|
||||
"{ _id: 0, type: \"chocolate\", orderDate: { $date : \"2020-05-18T14:10:30Z\" }, state: \"CA\", price: 13, quantity: 120 }"),
|
||||
Document.parse(
|
||||
"{ _id: 1, type: \"chocolate\", orderDate: { $date : \"2021-03-20T11:30:05Z\"}, state: \"WA\", price: 14, quantity: 140 }"),
|
||||
Document.parse(
|
||||
"{ _id: 2, type: \"vanilla\", orderDate: { $date : \"2021-01-11T06:31:15Z\"}, state: \"CA\", price: 12, quantity: 145 }"),
|
||||
Document.parse(
|
||||
"{ _id: 3, type: \"vanilla\", orderDate: { $date : \"2020-02-08T13:13:23Z\"}, state: \"WA\", price: 13, quantity: 104 }"),
|
||||
Document.parse(
|
||||
"{ _id: 4, type: \"strawberry\", orderDate: { $date : \"2019-05-18T16:09:01Z\"}, state: \"CA\", price: 41, quantity: 162 }"),
|
||||
Document.parse(
|
||||
"{ _id: 5, type: \"strawberry\", orderDate: { $date : \"2019-01-08T06:12:03Z\"}, state: \"WA\", price: 43, quantity: 134 }"));
|
||||
|
||||
collection.insertMany(source);
|
||||
return "OK";
|
||||
});
|
||||
}
|
||||
|
||||
@lombok.Data
|
||||
static class CakeSale {
|
||||
|
||||
@Id Integer id;
|
||||
|
||||
String state;
|
||||
|
||||
@Field("orderDate") //
|
||||
Date date;
|
||||
|
||||
@Field("quantity") //
|
||||
Integer qty;
|
||||
|
||||
String type;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
/*
|
||||
* 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.aggregation;
|
||||
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
|
||||
import java.util.Date;
|
||||
|
||||
import org.bson.Document;
|
||||
import org.junit.jupiter.api.Test;
|
||||
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.SetWindowFieldsOperation.Windows;
|
||||
import org.springframework.data.mongodb.core.convert.MappingMongoConverter;
|
||||
import org.springframework.data.mongodb.core.convert.NoOpDbRefResolver;
|
||||
import org.springframework.data.mongodb.core.convert.QueryMapper;
|
||||
import org.springframework.data.mongodb.core.mapping.Field;
|
||||
import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
|
||||
import org.springframework.lang.Nullable;
|
||||
|
||||
/**
|
||||
* @author Christoph Strobl
|
||||
*/
|
||||
public class SetWindowFieldsOperationUnitTests {
|
||||
|
||||
@Test // GH-3711
|
||||
void rendersTargetFieldNamesCorrectly() {
|
||||
|
||||
SetWindowFieldsOperation setWindowFieldsOperation = SetWindowFieldsOperation.builder() //
|
||||
.partitionByField("state") // resolves to field ref "$state"
|
||||
.sortBy(Sort.by(Direction.ASC, "date")) // resolves to "orderDate"
|
||||
.output(AccumulatorOperators.valueOf("qty").sum()) // resolves to "$quantity"
|
||||
.within(Windows.documents().fromUnbounded().toCurrent().build()) //
|
||||
.as("cumulativeQuantityForState") //
|
||||
.build(); //
|
||||
|
||||
Document document = setWindowFieldsOperation.toDocument(contextFor(CakeSale.class));
|
||||
assertThat(document).isEqualTo(Document.parse(
|
||||
"{ $setWindowFields: { partitionBy: \"$state\", sortBy: { orderDate: 1 }, output: { cumulativeQuantityForState: { $sum: \"$quantity\", window: { documents: [ \"unbounded\", \"current\" ] } } } } }"));
|
||||
}
|
||||
|
||||
@Test // GH-3711
|
||||
void exposesTargetFieldNames() {
|
||||
|
||||
SetWindowFieldsOperation setWindowFieldsOperation = SetWindowFieldsOperation.builder() //
|
||||
.output(AccumulatorOperators.valueOf("qty").sum()) // resolves to "$quantity"
|
||||
.within(Windows.documents().fromUnbounded().toCurrent().build()) //
|
||||
.as("f1") //
|
||||
.output(AccumulatorOperators.valueOf("qty").avg()) // resolves to "$quantity"
|
||||
.within(Windows.documents().from(-1).to(0).build()) //
|
||||
.as("f2") //
|
||||
.build(); //
|
||||
|
||||
assertThat(setWindowFieldsOperation.getFields()).map(ExposedField::getName).containsExactly("f1", "f2");
|
||||
}
|
||||
|
||||
@Test // GH-3711
|
||||
void rendersMuiltipleOutputFields() {
|
||||
|
||||
SetWindowFieldsOperation setWindowFieldsOperation = SetWindowFieldsOperation.builder() //
|
||||
.output(AccumulatorOperators.valueOf("qty").sum()) // resolves to "$quantity"
|
||||
.within(Windows.documents().fromUnbounded().toCurrent().build()) //
|
||||
.as("f1") //
|
||||
.output(AccumulatorOperators.valueOf("qty").avg()) // resolves to "$quantity"
|
||||
.within(Windows.documents().from(-1).to(0).build()) //
|
||||
.as("f2") //
|
||||
.build(); //
|
||||
|
||||
Document document = setWindowFieldsOperation.toDocument(contextFor(CakeSale.class));
|
||||
assertThat(document).isEqualTo(Document.parse(
|
||||
"{ $setWindowFields: { output: { f1 : { $sum: \"$quantity\", window: { documents: [ \"unbounded\", \"current\" ] } }, f2 : { $avg: \"$quantity\", window: { documents: [ -1, 0 ] } } } } }"));
|
||||
}
|
||||
|
||||
private static AggregationOperationContext contextFor(@Nullable Class<?> type) {
|
||||
|
||||
if (type == null) {
|
||||
return Aggregation.DEFAULT_CONTEXT;
|
||||
}
|
||||
|
||||
MappingMongoConverter mongoConverter = new MappingMongoConverter(NoOpDbRefResolver.INSTANCE,
|
||||
new MongoMappingContext());
|
||||
mongoConverter.afterPropertiesSet();
|
||||
|
||||
return new TypeBasedAggregationOperationContext(type, mongoConverter.getMappingContext(),
|
||||
new QueryMapper(mongoConverter));
|
||||
}
|
||||
|
||||
static class CakeSale {
|
||||
|
||||
String state;
|
||||
|
||||
@Field("orderDate") Date date;
|
||||
|
||||
@Field("quantity") Integer qty;
|
||||
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user