Java 类com.mongodb.AggregationOutput 实例源码

/**
 * 
 * @return number of total discovered links
 */
public Double getNumberOfDiscoveredLinks() {
    Double numberOfTriples = 0.0;
    try {
        DBCollection collection = DBSuperClass2.getDBInstance().getCollection(LinksetDB.COLLECTION_NAME);

        BasicDBObject groupFields = new BasicDBObject("_id", null);

        groupFields.append("sum", new BasicDBObject("$sum", "$links"));

        DBObject group = new BasicDBObject("$group", groupFields);

        // run aggregation
        List<DBObject> pipeline = Arrays.asList(group);
        AggregationOutput output = collection.aggregate(pipeline);

        for (DBObject result : output.results()) {
            numberOfTriples = Double.valueOf(result.get("sum").toString());
        }

    } catch (Exception e) {
        e.printStackTrace();
    }
    return numberOfTriples;
}

@Override public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
    Class<?>[] argumentsTypes, Object ret) throws Throwable {
    AbstractSpan activeSpan = ContextManager.activeSpan();
    CommandResult cresult = null;
    if (ret instanceof WriteResult) {
        WriteResult wresult = (WriteResult)ret;
        cresult = wresult.getCachedLastError();
    } else if (ret instanceof AggregationOutput) {
        AggregationOutput aresult = (AggregationOutput)ret;
        cresult = aresult.getCommandResult();
    }
    if (null != cresult && !cresult.ok()) {
        activeSpan.log(cresult.getException());
    }
    ContextManager.stopSpan();
    return ret;
}

private static Iterator<DBObject> setUpPipelineSample( String query, int numDocsToSample, DBCollection collection )
  throws KettleException {

  query = query + ", {$limit : " + numDocsToSample + "}"; //$NON-NLS-1$ //$NON-NLS-2$
  List<DBObject> samplePipe = jsonPipelineToDBObjectList( query );

  DBObject first = samplePipe.get( 0 );
  DBObject[] remainder = new DBObject[ samplePipe.size() - 1 ];
  for ( int i = 1; i < samplePipe.size(); i++ ) {
    remainder[ i - 1 ] = samplePipe.get( i );
  }

  AggregationOutput result = collection.aggregate( first, remainder );

  return result.results().iterator();
}

@Test public void testPipelineQueryIsLimited() throws KettleException, MongoDbException {
  setupPerform();

  AggregationOutput aggOutput = mock( AggregationOutput.class );
  Iterable<DBObject> results = mock( Iterable.class );
  when( aggOutput.results() ).thenReturn( results );
  when( results.iterator() ).thenReturn( mock( Iterator.class ) );

  String query = "{$sort : 1}";
  DBObject firstOp = (DBObject) JSON.parse( query );
  DBObject[] remainder = { new BasicDBObject( "$limit", NUM_DOCS_TO_SAMPLE ) };
  when( collection.aggregate( firstOp, remainder ) )
      .thenReturn( aggOutput );

  discoverFields.discoverFields( new MongoProperties.Builder(), "mydb", "mycollection", query, "", true,
      NUM_DOCS_TO_SAMPLE, inputMeta );

  verify( collection ).aggregate( firstOp, remainder );
}

@Override
protected List<UPSSubscription> getSubscriptions(String uri) {
    List<UPSSubscription> subscriptions = new ArrayList<>();

    List<String> listenerPaths = generatePaths(uri);

    DBObject match = new BasicDBObject("$match", new BasicDBObject("subscriptions.resource-path", new BasicDBObject("$in", listenerPaths)));
    DBObject unwind = new BasicDBObject("$unwind", "$subscriptions");

    // first perform a match to get the object which contains a subscription we want [uses the index]
    // then unwind to get the individual subscriptions
    // then match again to get only the subscriptions we want.
    // NOTE: the first step is not redundant, we need to first narrow down using an index (for performance) before unwinding and ultimately getting only the results we want.
    AggregationOutput aggregate = collection.aggregate(match, unwind, match);
    for (DBObject dbObject : aggregate.results()) {
        UPSSubscription subscription = UPSSubscription.create((DBObject) dbObject.get("subscriptions"));
        if (subscription != null) {
            subscriptions.add(subscription);
        }
    }

    return subscriptions;
}

@Override
public Map<String, Id> fetchAllArticleNumbers() {
    Map<String, Id> articleNumbers = new LinkedHashMap<String, Id>();

    Id attributeId = Attributes.getAttributeId(Product.class, "article_number");

    // build the $projection operation
    DBObject fields = new BasicDBObject("attributes.attr_id", 1);
    fields.put("attributes.val.val", 1);
    DBObject project = new BasicDBObject("$project", fields);

    DBObject unwind = new BasicDBObject("$unwind", "$attributes");

    // create our pipeline operations, first with the $match
    DBObject match = new BasicDBObject("$match", new BasicDBObject("attributes.attr_id", attributeId));

    // build the $projection operation
    DBObject project2 = new BasicDBObject("$project", new BasicDBObject("an", "$attributes.val.val"));

    List<DBObject> pipeline = Arrays.asList(project, unwind, match, project2);

    AggregationOutput output = db(Product.class).getCollection(COLLECTION_NAME).aggregate(pipeline);

    for (DBObject result : output.results()) {
        if (result.get("an") != null) {
            articleNumbers.put(String.valueOf(((BasicDBList) result.get("an")).get(0)).trim(),
                Id.valueOf(result.get("_id")));
        }
    }

    return articleNumbers;
}

@Override
public Map<Id, String> fetchIdArticleNumberMap() {
    Map<Id, String> articleNumbers = new LinkedHashMap<Id, String>();

    Id attributeId = Attributes.getAttributeId(Product.class, "article_number");

    // build the $projection operation
    DBObject fields = new BasicDBObject("attributes.attr_id", 1);
    fields.put("attributes.val.val", 1);
    DBObject project = new BasicDBObject("$project", fields);

    DBObject unwind = new BasicDBObject("$unwind", "$attributes");

    // create our pipeline operations, first with the $match
    DBObject match = new BasicDBObject("$match", new BasicDBObject("attributes.attr_id", attributeId));

    // build the $projection operation
    DBObject project2 = new BasicDBObject("$project", new BasicDBObject("an", "$attributes.val.val"));

    List<DBObject> pipeline = Arrays.asList(project, unwind, match, project2);

    AggregationOutput output = db(Product.class).getCollection(COLLECTION_NAME).aggregate(pipeline);

    for (DBObject result : output.results()) {
        if (result.get("an") != null) {
            articleNumbers.put(Id.valueOf(result.get("_id")),
                String.valueOf(((BasicDBList) result.get("an")).get(0)).trim());
        }
    }

    return articleNumbers;
}

@Override
public List<String> getSuggestions(Id attributeId, String collectionName, String lang, String query) {
    DBObject val = new BasicDBObject("$elemMatch",
        new BasicDBObject("val", new BasicDBObject("$regex", "^" + query + ".*")));

    DBObject elemMathContent = new BasicDBObject("attr_id", attributeId);
    elemMathContent.put("val", val);

    DBObject elemMatch = new BasicDBObject("$elemMatch", elemMathContent);

    List<DBObject> alls = new ArrayList<>();
    alls.add(elemMatch);

    DBObject match = new BasicDBObject("$match", new BasicDBObject("attributes", new BasicDBObject("$all", alls)));

    DBObject matchAdditionalContent = new BasicDBObject("attributes.attr_id", attributeId);
    if (lang != null && !lang.isEmpty()) {
        matchAdditionalContent.put("attributes.val.l", lang);
    }
    matchAdditionalContent.put("attributes.val.val", new BasicDBObject("$regex", "^" + query + ".*"));

    DBObject matchAdditional = new BasicDBObject("$match", matchAdditionalContent);
    AggregationOutput output = ((DB) connections.getFirstConnection("mongodb")).getCollection(collectionName)
        .aggregate(match, new BasicDBObject("$unwind", "$attributes"),
            new BasicDBObject("$unwind", "$attributes.val"), matchAdditional,
            new BasicDBObject("$group", new BasicDBObject("_id", "$attributes.val.val")),
            new BasicDBObject("$sort", new BasicDBObject("_id", 1)));

    List<String> suggestions = new ArrayList<>();

    for (DBObject result : output.results()) {
        suggestions.add((String) result.get("_id"));
    }

    return suggestions;
}

@Test
public void testDirect() throws Exception {
    Command cmd = this.utility.parseCommand("SELECT * FROM Customers");
    MongoDBConnection connection = Mockito.mock(MongoDBConnection.class);
    ExecutionContext context = Mockito.mock(ExecutionContext.class);
    DBCollection dbCollection = Mockito.mock(DBCollection.class);
    DB db = Mockito.mock(DB.class);
    Mockito.stub(db.getCollection("MyTable")).toReturn(dbCollection);

    Mockito.stub(db.collectionExists(Mockito.anyString())).toReturn(true);
    Mockito.stub(connection.getDatabase()).toReturn(db);

    AggregationOutput output = Mockito.mock(AggregationOutput.class);
    Mockito.stub(output.results()).toReturn(new ArrayList<DBObject>());

    Mockito.stub(dbCollection.aggregate(Mockito.any(DBObject.class),Mockito.any(DBObject.class))).toReturn(output);


    Argument arg = new Argument(Direction.IN, null, String.class, null);
    arg.setArgumentValue(new Literal("MyTable;{$match:{\"id\":\"$1\"}};{$project:{\"_m0\":\"$user\"}}", String.class));

    Argument arg2 = new Argument(Direction.IN, null, String.class, null);
    arg2.setArgumentValue(new Literal("foo", String.class));

    ResultSetExecution execution = this.translator.createDirectExecution(Arrays.asList(arg, arg2), cmd, context, this.utility.createRuntimeMetadata(), connection);
    execution.execute();

       List<DBObject> pipeline = TestMongoDBQueryExecution.buildArray(new BasicDBObject("$match", new BasicDBObject("id", "foo")), 
               new BasicDBObject("$project", new BasicDBObject("_m0", "$user")));        
       Mockito.verify(dbCollection).aggregate(Mockito.eq(pipeline), Mockito.any(AggregationOptions.class));

}

public Iterator<BasicDBObject> aggregate(String description, List<DBObject> pipeline) {
  if (log.isDebugEnabled()) {
    log.debug("--"+dbCollection.getName()+"-> "+description+" q="+toString(pipeline));
  }
  AggregationOutput aggregationOutput = dbCollection.aggregate(pipeline);
  return new LoggingIterator(this, aggregationOutput.results().iterator());
}

/**
 * 
 * @return number of total triples read
 */
public Double getNumberOfTriples() {
    Double numberOfTriples = 0.0;
    try {
        DBCollection collection = DBSuperClass2.getDBInstance().getCollection(DistributionDB.COLLECTION_NAME);

        BasicDBObject select = new BasicDBObject("$match",
                new BasicDBObject(DistributionDB.SUCCESSFULLY_DOWNLOADED, true));

        BasicDBObject groupFields = new BasicDBObject("_id", null);

        groupFields.append("sum", new BasicDBObject("$sum", "$triples"));

        DBObject group = new BasicDBObject("$group", groupFields);

        // run aggregation
        List<DBObject> pipeline = Arrays.asList(select, group);
        AggregationOutput output = collection.aggregate(pipeline);

        for (DBObject result : output.results()) {
            numberOfTriples = Double.valueOf(result.get("sum").toString());
        }

    } catch (Exception e) {
        e.printStackTrace();
    }
    return numberOfTriples;
}

private void serializeLoadTimes(JsonGenerator jgen) throws IOException,
        JsonProcessingException {
    jgen.writeArrayFieldStart("LoadTimes");
    /**
     * db.hubs.aggregate([ {$project: { "loadtimes" : 1}}, {$unwind :
     * "$loadtimes"}, {$project: { "truckType" : "$loadtimes.truckType",
     * "loadTime" : "$loadtimes.loadTime"}}])
     */

    AggregationOutput agg = getDB().getCollection("hubs").aggregate(
            new BasicDBObject().append("$project",
                    new BasicDBObject().append("loadtimes", 1)),
            new BasicDBObject().append("$unwind", "$loadtimes"),
            new BasicDBObject().append(
                    "$project",
                    new BasicDBObject().append("truckType",
                            "$loadtimes.truckType").append("loadTime",
                            "$loadtimes.loadTime")));

    for (DBObject obj : agg.results()) {
        jgen.writeStartObject();
        jgen.writeStringField("hub", obj.get("_id").toString());
        jgen.writeStringField("truckType", obj.get("truckType").toString());
        jgen.writeNumberField("loadTime",
                ((Number) obj.get("loadTime")).intValue());
        jgen.writeEndObject();
    }
    jgen.writeEndArray();
}

private void serializeRoutes(JsonGenerator jgen) throws IOException,
        JsonProcessingException {
    jgen.writeArrayFieldStart("Routes");
    /**
     * db.hubs.aggregate([ {$project: { "routes" : 1}}, {$unwind :
     * "$routes"}, {$project: { "spoke" : "$routes.spoke", "distance" :
     * "$routes.distance"}}])
     */

    AggregationOutput agg = getDB().getCollection("hubs").aggregate(
            new BasicDBObject().append("$project",
                    new BasicDBObject().append("routes", 1)),
            new BasicDBObject().append("$unwind", "$routes"),
            new BasicDBObject().append("$project",
                    new BasicDBObject().append("spoke", "$routes.spoke")
                            .append("distance", "$routes.distance")));

    for (DBObject obj : agg.results()) {
        jgen.writeStartObject();
        jgen.writeStringField("spoke", obj.get("spoke").toString());
        jgen.writeStringField("hub", obj.get("_id").toString());
        jgen.writeNumberField("distance",
                ((Number) obj.get("distance")).intValue());
        jgen.writeEndObject();
    }
    jgen.writeEndArray();
}

@Override
public Iterable<DBObject> results(){
    if(options == null){
        AggregationOutput output = coll.aggregate(pipeline);
        return output.results();
    }else{
        final Iterator<DBObject> it = coll.aggregate(pipeline, options);
        return new Iterable<DBObject>() {
            @Override
            public Iterator<DBObject> iterator() {
                return it;
            }
        };
    }
}

public Iterator<BasicDBObject> aggregate(String description, List<DBObject> pipeline) {
  if (log.isDebugEnabled()) {
    log.debug("--"+dbCollection.getName()+"-> "+description+" q="+toString(pipeline));
  }
  AggregationOutput aggregationOutput = dbCollection.aggregate(pipeline);
  return new LoggingIterator(this, aggregationOutput.results().iterator());
}

/**
 * Convert aggregation output to BasicDBList.
 *
 * @param aggregationOutput aggregation output
 * @return BasicDBList
 */
public static BasicDBList getAsBasicDBList(AggregationOutput aggregationOutput) {
    Validate.notNull(aggregationOutput, "aggregation output cannot be null");
    BasicDBList result = new BasicDBList();
    result.addAll(Lists.newArrayList(aggregationOutput.results()));
    return result;
}

@Override
<T> List<T> queryForList(QueryStatement queryStatement, Class<T> type) {
    DBCollection dbCollection = getDbCollection(queryStatement.getCollectionName());
    BasicDBList query = (BasicDBList) JSON_TO_DB_OBJECT_MARSHALLER.marshall(queryStatement.getPreparedQuery(),
            queryStatement.getParameters());
    AggregationOutput aggregationOutput = performAggregationQuery(dbCollection, query);
    BasicDBList source = getAsBasicDBList(aggregationOutput);
    List<T> result = convertList(type, source, queryStatement.getConverterClass(),
            queryStatement.getConverterMethod());
    return result != null ? result : Lists.<T>newArrayList();
}

@Override
<T> T queryForObject(QueryStatement queryStatement, Class<T> type) {
    DBCollection dbCollection = getDbCollection(queryStatement.getCollectionName());
    BasicDBList query = (BasicDBList) JSON_TO_DB_OBJECT_MARSHALLER.marshall(queryStatement.getPreparedQuery(),
            queryStatement.getParameters());
    AggregationOutput aggregationOutput = performAggregationQuery(dbCollection, query);
    BasicDBList result = getAsBasicDBList(aggregationOutput);
    return convertOne(type, result, queryStatement.getConverterClass(), queryStatement.getConverterMethod());
}

public String reserved(JID jid) {
    // {"$match":{"jid":group.bare}}, {"$unwind":"$affiliations"}, {"$match":{"affiliations.jid":jid.bare}}, {"$project":{"nick":"$affiliations.nick"}}
    AggregationOutput output = MongoRoomBuilder.this.config.collection().aggregate(BasicDBObjectBuilder.start("$match", BasicDBObjectBuilder.start(Dictionary.FIELD_JID, this.group.asStringWithBare()).get()).get(), MongoRoomBuilder.this.unwind, BasicDBObjectBuilder.start("$match", BasicDBObjectBuilder.start(Dictionary.FIELD_AFFILIATIONS + "." + Dictionary.FIELD_JID, jid.asStringWithBare()).get()).get(), MongoRoomBuilder.this.project);
    @SuppressWarnings("deprecation")
    List<?> result = MongoUtils.asList(output.getCommandResult(), Dictionary.FIELD_RESULT);
    return result.isEmpty() ? null : MongoUtils.asString(DBObject.class.cast(result.get(0)), Dictionary.FIELD_NICK);
}

@Override
public Relation ourRelation(JID from, JID to) {
    AggregationOutput output = this.config.collection().aggregate(this.buildMatcher(to), this.unwindRoles, this.unwindAffiliation, BasicDBObjectBuilder.start().add("$match", BasicDBObjectBuilder.start().add(Dictionary.FIELD_ROLES + "." + Dictionary.FIELD_JID, from.asStringWithBare()).add(Dictionary.FIELD_ROLES + "." + Dictionary.FIELD_RESOURCE, from.resource()).get()).get(), this.projectRelation, this.match, this.sort, this.limit);
    @SuppressWarnings("deprecation")
    List<?> result = MongoUtils.asList(output.getCommandResult(), Dictionary.FIELD_RESULT);
    return result.isEmpty() ? new NoneRelation(from, to, this.affiliation(from, to)) : new MongoRelation(DBObject.class.cast(result.get(0)));
}

@Override
public Set<JID> whoSubscribedMe(JID from) {
    AggregationOutput output = this.config.collection().aggregate(this.buildMatcher(from), this.projectRoles, this.unwindRoles, this.groupSubscribe);
    @SuppressWarnings("deprecation")
    List<?> result = MongoUtils.asList(output.getCommandResult(), Dictionary.FIELD_RESULT);
    return result.isEmpty() ? this.jidset : new JIDGroup(MongoUtils.asList(DBObject.class.cast(result.get(0)), Dictionary.FIELD_ROLES));
}

@Override
public JIDs mapping(JID group) {
    // {"$match":{"jid":group.bare}}, {"$unwind":"$roles"}, {"$match":{"roles.nick":Xxx}}, {"$project":{"roles":"$roles"}}, {"$group":{"_id":"$roles.jid","resource":{"$push":"$roles.resource"}}}
    AggregationOutput output = this.config.collection().aggregate(this.buildMatcher(group), this.unwindRoles, BasicDBObjectBuilder.start().add("$match", BasicDBObjectBuilder.start(Dictionary.FIELD_ROLES + "." + Dictionary.FIELD_NICK, group.resource()).get()).get(), this.projectRoles, this.groupMapping);
    @SuppressWarnings("deprecation")
    List<?> result = MongoUtils.asList(output.getCommandResult(), Dictionary.FIELD_RESULT);
    return result.isEmpty() ? this.jids : this.extract(DBObject.class.cast(result.get(0)));
}

public Set<Relation> myRelations(JID from, String affiliation) {
    // {"$match":{"jid":group.bare}}, {"$unwind":"$affiliations"}, {"$match":{"affiliations.affiliation":Xxx}}, {"$project":{"affiliation":"$affiliations"}}
    AggregationOutput output = super.config.collection().aggregate(super.buildMatcher(from), super.unwindAffiliation, BasicDBObjectBuilder.start("$match", BasicDBObjectBuilder.start(Dictionary.FIELD_AFFILIATIONS + "." + Dictionary.FIELD_AFFILIATION, affiliation).get()).get(), this.projectAffiliation);
    @SuppressWarnings("deprecation")
    List<?> result = MongoUtils.asList(output.getCommandResult(), Dictionary.FIELD_RESULT);
    return result.isEmpty() ? super.relations : new AffiliationRelations(result);
}

public Set<Relation> myRelations(JID from, String role) {
    // {"$match":{"jid":group.bare}}, {"$unwind":"$roles"}, {"$match":{"roles.role":Xxx}}, {"$group":{"_id":{"jid":"$jid","creator":"$creator","affiliations":"$affiliations"},"roles":{"$addToSet":"$roles"}}}, {"$project":{"jid":"$_id.jid","creator":"$_id.creator","affiliations":"$_id.affiliations","roles":"$roles"}}
    AggregationOutput output = super.config.collection().aggregate(this.buildMatcher(from), super.unwindRoles, BasicDBObjectBuilder.start().add("$match", BasicDBObjectBuilder.start(Dictionary.FIELD_ROLES + "." + Dictionary.FIELD_ROLE, role).get()).get(), this.group, this.projectRole);
    @SuppressWarnings("deprecation")
    List<?> result = MongoUtils.asList(output.getCommandResult(), Dictionary.FIELD_RESULT);
    return result.isEmpty() ? this.relations : new MongoRelations(DBObject.class.cast(result.get(0)));
}

public String getTopGuilds(final String matchid) {
    final WvwMatch wvwmatch = findWvwMatch(matchid);
    if(wvwmatch == null){
        return "[]";
    }

    final DBObject timerange = new BasicDBObject("$gt", wvwmatch.getStart_time());
    timerange.put("$lt", wvwmatch.getEnd_time());
    final DBObject matchFields = new BasicDBObject("timestamp", timerange);
    matchFields.put("match_id", matchid);
    final DBObject match = new BasicDBObject("$match", matchFields);

    final DBObject projectFields = new BasicDBObject("guild_id", 1);
    projectFields.put("holdtime", 1);
    projectFields.put("_id", 0);
    final DBObject project = new BasicDBObject("$project", projectFields);

    final DBObject groupFields = new BasicDBObject("_id", "$guild_id");
    groupFields.put("holdtime", new BasicDBObject("$sum", "$holdtime"));
    groupFields.put("count", new BasicDBObject("$sum", 1));
    final DBObject group = new BasicDBObject("$group", groupFields);

    final DBObject sortFields = new BasicDBObject("count", -1);
    sortFields.put("holdtime", -1);
    final DBObject sort = new BasicDBObject("$sort", sortFields);

    final DBObject limit = new BasicDBObject("$limit", 10);

    final AggregationOutput output = wvwguildstatistics.getDbCollection().aggregate(match, project, group, sort, limit);
    return output.results().toString();
}

public Iterable<DBObject> getAggregationQuery(String key, int min, int max) {
    DBObject match = new BasicDBObject(key, new BasicDBObject("$gt", min).append("$lt", max));
    DBObject matchOp = new BasicDBObject("$match", match);
    DBCollection fullData = mc.getCollection(type);
    DBObject groupOp = setupGroupOperation();
    AggregationOutput ao = fullData.aggregate(matchOp, groupOp);
    return ao.results();
}

public TimeSeries getTimeSerie(DataType t, DBObject groupOp, String groupNameX, String groupNameY, String cName, Boolean useMovingAverage) {
    long begin = new Date().getTime();

    // Query fetches collection t
    DBCollection c = getCollection(t);
    AggregationOutput ao = c.aggregate(groupOp);
    TimeSeries timeserie = new TimeSeries(cName);

    HashMap<Integer, Integer> valueMap = new HashMap<Integer, Integer>();
    List<Integer> xVals = new ArrayList<Integer>();
    for (DBObject dbo : ao.results()) {
        int x = (Integer) dbo.get(groupNameX);
        int y = (Integer) dbo.get(groupNameY);
        xVals.add(x);
        valueMap.put(x, y);
    }

    int min = Collections.min(xVals);
    int max = Collections.max(xVals);
    for (int i = min; i <= max; i += 60) {
        timeserie.add(new Minute(new Date(i * 1000L)), valueMap.get(i));
    }

    if (useMovingAverage) {
        timeserie = MovingAverage.createMovingAverage(timeserie, cName, 50, 100);
    }

    logger.info("Fetched slider backgroud data, DataType:" + t.toString() + ", query took " + (new Date().getTime() - begin) + " ms");

    return timeserie;
}

private BasicDBList aggregate(RequestContext ctx) {
    BasicDBList queryObject = new BasicDBList();
    if (ctx.resourceParams() != null && ctx.resourceParams().contains("q")) {
        String queryString = ctx.resourceParams().value("q");
        DBObject paramObject = (DBObject) JSON.parse(queryString);

        if (paramObject instanceof BasicDBList) {
            queryObject = (BasicDBList) paramObject;
        } else {
            queryObject.add(paramObject);
        }
    }

    DBCollection dbCollection = parent().getDBCollection();

    try {
        BasicDBList result = new BasicDBList();
        AggregationOutput output = dbCollection.aggregate(
                (DBObject) queryObject.remove(0),
                queryObject.toArray(new DBObject[queryObject.size()]));
        for (DBObject dbObject : output.results()) {
            result.add(dbObject);
        }

        return result;

    } catch (Exception e) {
        logger().error("", e);
        throw new RuntimeException("Aggregation query failed: ", e);
    }
}

private MongoStatistics calculateAndSaveStatistics(StatisticsFilter request, MongoStatisticKey key) {
    AggregationOutput aggregate = aggregate(matches(request),
            project(),
            unwind(),
            group());
    List<MongoStatistic> statistics =
                parseStatistics(aggregate.results());
    MongoStatistics v = new MongoStatistics(key, statistics);
    this.dao.save(v);
    return v;
}

private AggregationOutput aggregate(DBObject firstOp,
                                    DBObject... additionalOps) {
    AggregationOutput result = mongoDB.getDatastore()
            .getCollection(MongoMeasurement.class)
            .aggregate(firstOp, additionalOps);
    return result;
}

public List<ObjectId> getIds(User user) {
        DBObject group = MongoUtils.group(new BasicDBObject(ID, MongoUtils.valueOf(MongoMeasurement.SENSOR, ID)));
        DBObject match = MongoUtils.match(MongoMeasurement.USER, ref(user));
        AggregationOutput result = getMongoDB().getDatastore()
                .getCollection(MongoMeasurement.class).aggregate(match, group);
        return StreamSupport.stream(result.results().spliterator(), false)
                .map(x -> (ObjectId) x.get(ID))
//                .map(x -> new Key<>(MongoSensor.class, x))
                .collect(Collectors.toList());
    }

List<Key<MongoTrack>> getTrackKeysByBbox(MeasurementFilter filter) {
    ArrayList<DBObject> filters = new ArrayList<>(4);
    if (filter.hasSpatialFilter()) {
        SpatialFilter sf = filter.getSpatialFilter();
        if (sf.getOperator()==SpatialFilterOperator.BBOX){
            filters.add(matchGeometry(filter.getSpatialFilter().getGeom()));
        }
        //TODO add further spatial filters
    }
    if (filter.hasUser()) {
        filters.add(matchUser(filter.getUser()));
    }
    if (filter.hasTrack()) {
        filters.add(matchTrack(filter.getTrack()));
    }
    if (filter.hasTemporalFilter()) {
        filters.add(matchTime(filter.getTemporalFilter()));
    }

    final AggregationOutput out;
    if (filters.isEmpty()) {
        out = aggregate(project(), group());
    } else {
        int size = filters.size();
        if (size == 1) {
            out = aggregate(filters.get(0), project(), group());
        } else {
            DBObject first = filters.get(0);
            DBObject[] other = new DBObject[size + 1];
            for (int i = 1; i < size; ++i) {
                other[i - 1] = filters.get(i);
            }
            other[other.length - 2] = project();
            other[other.length - 1] = group();
            out = aggregate(first, other);
        }
    }
    return toKeyList(out.results());
}

private AggregationOutput aggregate(DBObject firstOp,
                                    DBObject... additionalOps) {
    AggregationOutput result = mongoDB.getDatastore()
            .getCollection(MongoMeasurement.class)
            .aggregate(firstOp, additionalOps);
    return result;
}

/**
 * Returns the point in time when followt started monitoring the given user.
 * @param followee the user for which the beginning of time should be returned
 * @return point in time when the first scan of that user started
 */
public Date beginningOfTime (int followee) {
    AggregationOutput agr = fhistory.aggregate(
        new BasicDBObject("$match",
            new BasicDBObject ("followee",followee)),
        new BasicDBObject("$group",
            new BasicDBObject("_id",null)
                      .append("beginning_of_time", 
                              new BasicDBObject("$min","$start"))));
    DBObject result = agr.results().iterator().next();
    return (Date)result.get("beginning_of_time");
}

public AggregationOutputImpl(AggregationOutput ao) {
    this.ao=ao;
}

public AggregationOutput getAggregationOutput(){
    return ao;
}

public Set<Relation> ourRelations(JID from, JID to) {
    AggregationOutput output = this.config.collection().aggregate(this.buildMatcher(to), this.unwindRoles, BasicDBObjectBuilder.start().add("$match", BasicDBObjectBuilder.start().add(Dictionary.FIELD_ROLES + "." + Dictionary.FIELD_JID, from.asStringWithBare()).get()).get(), this.groupRelations, this.projectRelations);
    @SuppressWarnings("deprecation")
    List<?> result = MongoUtils.asList(output.getCommandResult(), Dictionary.FIELD_RESULT);
    return result.isEmpty() ? this.relations : new MongoRelations(DBObject.class.cast(result.get(0)));
}

public AggregationOutput aggregate(final DBObject... ops) {
    this.log("aggregate", ops);
    return MongoProxyConfig.this.collection.aggregate(Arrays.asList(ops));
}

protected void initQuery() throws KettleException, MongoDbException {

    // close any previous cursor
    if ( data.cursor != null ) {
      data.cursor.close();
    }

    // check logging level and only set to false if
    // logging level at least detailed
    if ( log.isDetailed() ) {
      m_serverDetermined = false;
    }

    String query = environmentSubstitute( meta.getJsonQuery() );
    String fields = environmentSubstitute( meta.getFieldsName() );
    if ( Const.isEmpty( query ) && Const.isEmpty( fields ) ) {
      if ( meta.getQueryIsPipeline() ) {
        throw new KettleException( BaseMessages
            .getString( MongoDbInputMeta.PKG, "MongoDbInput.ErrorMessage.EmptyAggregationPipeline" ) ); //$NON-NLS-1$
      }

      data.cursor = data.collection.find();
    } else {

      if ( meta.getQueryIsPipeline() ) {
        if ( Const.isEmpty( query ) ) {
          throw new KettleException( BaseMessages
              .getString( MongoDbInputMeta.PKG, "MongoDbInput.ErrorMessage.EmptyAggregationPipeline" ) ); //$NON-NLS-1$
        }

        if ( meta.getExecuteForEachIncomingRow() && m_currentInputRowDrivingQuery != null ) {
          // do field value substitution
          query = fieldSubstitute( query, getInputRowMeta(), m_currentInputRowDrivingQuery );
        }

        logDetailed( BaseMessages.getString( PKG, "MongoDbInput.Message.QueryPulledDataFrom", query ) );

        List<DBObject> pipeline = MongodbInputDiscoverFieldsImpl.jsonPipelineToDBObjectList( query );
        DBObject firstP = pipeline.get( 0 );
        DBObject[] remainder = null;
        if ( pipeline.size() > 1 ) {
          remainder = new DBObject[pipeline.size() - 1];
          for ( int i = 1; i < pipeline.size(); i++ ) {
            remainder[i - 1] = pipeline.get( i );
          }
        } else {
          remainder = new DBObject[0];
        }

        AggregationOutput result = data.collection.aggregate( firstP, remainder );
        data.m_pipelineResult = result.results().iterator();
      } else {
        if ( meta.getExecuteForEachIncomingRow() && m_currentInputRowDrivingQuery != null ) {
          // do field value substitution
          query = fieldSubstitute( query, getInputRowMeta(), m_currentInputRowDrivingQuery );

          fields = fieldSubstitute( fields, getInputRowMeta(), m_currentInputRowDrivingQuery );
        }

        logDetailed( BaseMessages.getString( PKG, "MongoDbInput.Message.ExecutingQuery", query ) );

        DBObject dbObject = (DBObject) JSON.parse( Const.isEmpty( query ) ? "{}" //$NON-NLS-1$
            : query );
        DBObject dbObject2 = (DBObject) JSON.parse( fields );
        data.cursor = data.collection.find( dbObject, dbObject2 );
      }
    }
  }

/**
 * Perform aggregation query.
 *
 * @param dbCollection db collection
 * @param operators    operators
 * @return {@link AggregationOutput}
 */
protected AggregationOutput performAggregationQuery(DBCollection dbCollection, BasicDBList operators) {
    Validate.notNull(dbCollection, "dbCollection cannot be null");
    Validate.notEmpty(operators, "operators cannot be null or empty");
    DBObject firstOperator = (DBObject) operators.remove(FIRST_ELEMENT);
    return dbCollection.aggregate(firstOperator, operators.toArray(new DBObject[FIRST_ELEMENT]));
}