Using aggregation parameters with @Aggregation

[alturkovic]

I am trying to run the following MongoDB aggregation using Spring:

@Repository
public interface StockRepository extends MongoRepository<StockDocument, UUID> {
    @Aggregation(pipeline = """
        [
          {
            $match: {
              timestamp: {
                $gte: $?0,
                $lte: $?1
              }
            }
          },
          {
            $group: {
              _id: {
                $toDate: {
                  $subtract: [
                    { $toLong: '$timestamp' },
                    { $mod: [{ $toLong: '$timestamp' }, $?2] }
                  ]
                }
              },
              avgOpen: { $avg: '$open' },
              avgClose: { $avg: '$close' },
              minLow: {$min: '$low'},
              maxHigh: {$max: '$high'}
            }
          },
          {
            $sort: {
              _id: 1
            }
          }
        ]
    """)
    AggregationResults<Document> queryStockPrices(Instant from, Instant to, long intervalMs);
}

The query should find all documents in a time range and aggregate them using a specific time interval (the ?2 parameter).

When I run this pipeline directly on Mongo using specific dates and mod parameter, everything works as expected. Example:

     [
          {
            $match: {
              timestamp: {
                $gte: ISODate('2023-03-02T11:39:00.000+00:00'),
                $lte: ISODate('2023-03-02T11:55:00.000+00:00')
              }
            }
          },
          {
            $group: {
              _id: {
                $toDate: {
                  $subtract: [
                    { $toLong: '$timestamp' },
                    { $mod: [{ $toLong: '$timestamp' }, 60000] }
                  ]
                }
              },
              avgOpen: { $avg: '$open' },
              avgClose: { $avg: '$close' },
              minLow: {$min: '$low'},
              maxHigh: {$max: '$high'}
            }
          },
          {
            $sort: {
              _id: 1
            }
          }
        ]

But, when I run this using Spring, I get the following exception:

java.util.NoSuchElementException: null
    at java.base/java.util.LinkedHashMap$LinkedHashIterator.nextNode(LinkedHashMap.java:758) ~[na:na]
    at java.base/java.util.LinkedHashMap$LinkedKeyIterator.next(LinkedHashMap.java:778) ~[na:na]
    at org.springframework.data.mongodb.core.aggregation.AggregationOperation.getOperator(AggregationOperation.java:66) ~[spring-data-mongodb-4.2.5.jar:4.2.5]
    at org.springframework.data.mongodb.core.aggregation.AggregationPipeline.isOut(AggregationPipeline.java:165) ~[spring-data-mongodb-4.2.5.jar:4.2.5]
    at org.springframework.data.mongodb.core.aggregation.AggregationPipeline.verify(AggregationPipeline.java:107) ~[spring-data-mongodb-4.2.5.jar:4.2.5]
    at org.springframework.data.mongodb.core.aggregation.AggregationPipeline.toDocuments(AggregationPipeline.java:85) ~[spring-data-mongodb-4.2.5.jar:4.2.5]
    at org.springframework.data.mongodb.core.aggregation.Aggregation.toPipeline(Aggregation.java:757) ~[spring-data-mongodb-4.2.5.jar:4.2.5]
    at org.springframework.data.mongodb.core.AggregationUtil.createPipeline(AggregationUtil.java:98) ~[spring-data-mongodb-4.2.5.jar:4.2.5]
    at org.springframework.data.mongodb.core.MongoTemplate.doAggregate(MongoTemplate.java:2173) ~[spring-data-mongodb-4.2.5.jar:4.2.5]
    at org.springframework.data.mongodb.core.MongoTemplate.doAggregate(MongoTemplate.java:2148) ~[spring-data-mongodb-4.2.5.jar:4.2.5]
    at org.springframework.data.mongodb.core.MongoTemplate.aggregate(MongoTemplate.java:2142) ~[spring-data-mongodb-4.2.5.jar:4.2.5]
    at org.springframework.data.mongodb.core.MongoTemplate.aggregate(MongoTemplate.java:2008) ~[spring-data-mongodb-4.2.5.jar:4.2.5]
    at org.springframework.data.mongodb.core.MongoTemplate.aggregate(MongoTemplate.java:2002) ~[spring-data-mongodb-4.2.5.jar:4.2.5]
    at org.springframework.data.mongodb.repository.query.StringBasedAggregation.doExecute(StringBasedAggregation.java:126) ~[spring-data-mongodb-4.2.5.jar:4.2.5]
    at org.springframework.data.mongodb.repository.query.AbstractMongoQuery.execute(AbstractMongoQuery.java:119) ~[spring-data-mongodb-4.2.5.jar:4.2.5]
    at org.springframework.data.repository.core.support.RepositoryMethodInvoker.doInvoke(RepositoryMethodInvoker.java:170) ~[spring-data-commons-3.2.5.jar:3.2.5]
    at org.springframework.data.repository.core.support.RepositoryMethodInvoker.invoke(RepositoryMethodInvoker.java:158) ~[spring-data-commons-3.2.5.jar:3.2.5]
    at org.springframework.data.repository.core.support.QueryExecutorMethodInterceptor.doInvoke(QueryExecutorMethodInterceptor.java:164) ~[spring-data-commons-3.2.5.jar:3.2.5]
    at org.springframework.data.repository.core.support.QueryExecutorMethodInterceptor.invoke(QueryExecutorMethodInterceptor.java:143) ~[spring-data-commons-3.2.5.jar:3.2.5]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:184) ~[spring-aop-6.1.6.jar:6.1.6]
    at org.springframework.data.projection.DefaultMethodInvokingMethodInterceptor.invoke(DefaultMethodInvokingMethodInterceptor.java:70) ~[spring-data-commons-3.2.5.jar:3.2.5]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:184) ~[spring-aop-6.1.6.jar:6.1.6]
    at org.springframework.data.mongodb.repository.support.CrudMethodMetadataPostProcessor$CrudMethodMetadataPopulatingMethodInterceptor.invoke(CrudMethodMetadataPostProcessor.java:129) ~[spring-data-mongodb-4.2.5.jar:4.2.5]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:184) ~[spring-aop-6.1.6.jar:6.1.6]
    at org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:97) ~[spring-aop-6.1.6.jar:6.1.6]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:184) ~[spring-aop-6.1.6.jar:6.1.6]
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:223) ~[spring-aop-6.1.6.jar:6.1.6]
    at jdk.proxy2/jdk.proxy2.$Proxy96.queryStockPrices(Unknown Source) ~[na:na]
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:na]
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77) ~[na:na]
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:na]
    at java.base/java.lang.reflect.Method.invoke(Method.java:568) ~[na:na]
    at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:354) ~[spring-aop-6.1.6.jar:6.1.6]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:196) ~[spring-aop-6.1.6.jar:6.1.6]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163) ~[spring-aop-6.1.6.jar:6.1.6]
    at org.springframework.dao.support.PersistenceExceptionTranslationInterceptor.invoke(PersistenceExceptionTranslationInterceptor.java:137) ~[spring-tx-6.1.6.jar:6.1.6]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:184) ~[spring-aop-6.1.6.jar:6.1.6]
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:223) ~[spring-aop-6.1.6.jar:6.1.6]
    at jdk.proxy2/jdk.proxy2.$Proxy96.queryStockPrices(Unknown Source) ~[na:na]

I have tried wrapping the parameters, such as '$?0' but that did not work either, even though that is the format mentioned in @Aggregation JavaDoc.

What am I doing wrong? How do I pass these parameters correctly in my aggregation pipeline using Spring @Aggregation?

[christophstrobl]

Thank you for getting in touch. Please do not feed the entire pipeline as a string. Aggregation#pipeline accepts individual stages. Also, there is this extra $ before each parameter binding, that is pushing the reader into scanUnquotedString mode not suitable for data binding. You may want to use something like timestamp: { $gte: ?0, $lte: ?1 } to bind the given java.time input parameter.

[alturkovic]

Please do not feed the entire pipeline as a string. Aggregation#pipeline accepts individual stages.

My bad, I did not notice the pipeline accepts multiple strings, thanks!

Also, there is this extra $ before each parameter binding, that is pushing the reader into scanUnquotedString mode not suitable for data binding.

Thank you, that helped also, I got confused by the JavaDoc example:

  @Aggregation("{ '$group': { '_id' : '$lastname', names : { $addToSet : '$?0' } } }")
  List<PersonAggregate> groupByLastnameAnd(String property);

Maybe it is just me, but a simpler aggregation pipeline (not using the property, but actual values) in the JavaDoc might prevent someone in the future from doing the same mistake I did.

Sorry for the noise and thank you for your time.

[christophstrobl]

thank you for the feedback, glad it works for you now - yeah, makes sense to revisit the documentation.