/**
 * Takes `uploadID` and returns all bed names as a json format string
 * @param uploadID - the year that the data was uploaded
 * @return String representation of json with all bed names
 */
public String getGardenLocationsAsJson(String uploadID){
    AggregateIterable<Document> documents
            = plantCollection.aggregate(
            Arrays.asList(
                    Aggregates.match(eq("uploadID",uploadID)),//!! Order is important here
                    Aggregates.group("$gardenLocation"),Aggregates.sort(Sorts.ascending("_id"))
            ));

    List<Document> listDoc = new ArrayList<>();
    for (Document doc : documents) {
        listDoc.add(doc);
    }
    listDoc.sort(new bedComparator());

    return JSON.serialize(listDoc);
}

/**
 *
 * @return a date-sorted List of all the distinct uploadIds in the DB
 */
public static List<String> listUploadIds(MongoDatabase database) {
    MongoCollection<Document> plantCollection = database.getCollection("plants");

    AggregateIterable<Document> documents
            = plantCollection.aggregate(
            Arrays.asList(
                    Aggregates.group("$uploadId"),Aggregates.sort(Sorts.ascending("_id"))
            ));
    List<String> lst = new LinkedList<>();
    for(Document d: documents) {
        lst.add(d.getString("_id"));
    }
    return lst;
}

public int returnValuetoURL(String URL)
{
    Block<Document> printBlock = new Block<Document>() {
        @Override
        public void apply(final Document document) {
            System.out.println(document.toJson());
        }
    };
    MongoCollection<Document> collection = db.getCollection("ratings");

    collection.aggregate(
            Arrays.asList(
                    Aggregates.group("URL",Accumulators.avg("rating",1))))
            .forEach(printBlock);
    System.out.println(printBlock.toString());
    return 0;
}

@Test
public void ageCounts() {
    AggregateIterable<Document> documents
            = userDocuments.aggregate(
            Arrays.asList(
                    /*
                     * Groups data by the "age" field,and then counts
                     * the number of documents with each given age.
                     * This creates a new "constructed document" that
                     * has "age" as it's "_id",and the count as the
                     * "ageCount" field.
                     */
                    Aggregates.group("$age",Accumulators.sum("ageCount",1)),Aggregates.sort(Sorts.ascending("_id"))
            )
    );
    List<Document> docs = intoList(documents);
    assertEquals("Should be two distinct ages",2,docs.size());
    assertEquals(docs.get(0).get("_id"),25);
    assertEquals(docs.get(0).get("ageCount"),1);
    assertEquals(docs.get(1).get("_id"),37);
    assertEquals(docs.get(1).get("ageCount"),2);
}

@Test
public void averageAge() {
    AggregateIterable<Document> documents
            = userDocuments.aggregate(
                    Arrays.asList(
                            Aggregates.group("$company",Accumulators.avg("averageAge","$age")),Aggregates.sort(Sorts.ascending("_id"))
                    ));
    List<Document> docs = intoList(documents);
    assertEquals("Should be three companies",3,docs.size());

    assertEquals("Frogs,Inc.",docs.get(0).get("_id"));
    assertEquals(37.0,docs.get(0).get("averageAge"));
    assertEquals("IBM",docs.get(1).get("_id"));
    assertEquals(37.0,docs.get(1).get("averageAge"));
    assertEquals("UMM",docs.get(2).get("_id"));
    assertEquals(25.0,docs.get(2).get("averageAge"));
}

@Test
public void ageCounts() {
    AggregateIterable<Document> documents
            = userDocuments.aggregate(
            Arrays.asList(
                    /*
                     * Groups data by the "age" field,2);
}

@Test
public void averageAge() {
    AggregateIterable<Document> documents
            = userDocuments.aggregate(
                    Arrays.asList(
                            Aggregates.group("$company",docs.get(2).get("averageAge"));
}

@Test
public void ageCounts() {
    AggregateIterable<Document> documents
            = userDocuments.aggregate(
            Arrays.asList(
                    /*
                     * Groups data by the "age" field,2);
}

@Test
public void averageAge() {
    AggregateIterable<Document> documents
            = userDocuments.aggregate(
                    Arrays.asList(
                            Aggregates.group("$company",docs.get(2).get("averageAge"));
}

private static List<Bson> getServerQuery(Bson filter) {

    final List<Bson> pipeline = new ArrayList<>(6);
    if (filter != ALL) {
      pipeline.add(Aggregates.match(filter));
    }
    pipeline.add(Aggregates.unwind("$deployments",new UnwindOptions().preserveNullAndEmptyArrays(true)));
    pipeline.add(Aggregates.lookup(Collections.APPLICATIONS,"deployments.applicationId","id","applications"));
    pipeline.add(Aggregates.unwind("$applications",new UnwindOptions().preserveNullAndEmptyArrays(true)));
    pipeline.add(Aggregates.group(
      new Document().append("hostname","$hostname").append("environment","$environment"),new BsonField("fqdn",new Document("$first","$fqdn")),new BsonField("description","$description")),new BsonField("os","$os")),new BsonField("network","$network")),new BsonField("Meta","$Meta")),new BsonField("attributes","$attributes")),new BsonField("applications",new Document("$push","$applications")),new BsonField("deployments","$deployments"))));
    pipeline.add(Aggregates.sort(Sorts.ascending("_id")));
    return pipeline;
  }

/**
 * Get a json containing a list of commonNames sorted by common name
 * @param uploadID
 * @return
 */
public String getCommonNamesJSON(String uploadID){
    if (!ExcelParser.isValidUploadId(db,uploadID))
        return "null";

    AggregateIterable<Document> documents
            = plantCollection.aggregate(
            Arrays.asList(
                    Aggregates.match(eq("uploadId",//!! Order is important here
                    Aggregates.group("$commonName"),Aggregates.sort(Sorts.ascending("commonName"))
            ));
    return JSON.serialize(documents);
}

public String getGardenLocationsAsJson(String uploadID) {
    AggregateIterable<Document> documents
            = plantCollection.aggregate(
            Arrays.asList(
                    Aggregates.match(eq("uploadId",Aggregates.sort(Sorts.ascending("_id"))
            ));
    return JSON.serialize(documents);
}

/**
     *
     * @return a sorted JSON array of all the distinct uploadIds in the DB
     */
    public String listUploadIds() {
        AggregateIterable<Document> documents
                = plantCollection.aggregate(
                Arrays.asList(
                        Aggregates.group("$uploadId"),Aggregates.sort(Sorts.ascending("_id"))
                ));
        List<String> lst = new LinkedList<>();
        for(Document d: documents) {
            lst.add(d.getString("_id"));
        }
        return JSON.serialize(lst);
//        return JSON.serialize(plantCollection.distinct("uploadId","".getClass()));
    }

/**
     *
     * @return a sorted JSON array of all the distinct uploadIDs in plant collection of the DB
     */
    public List<String> listUploadIDs() {
        AggregateIterable<Document> documents
                = plantCollection.aggregate(
                Arrays.asList(
                        Aggregates.group("$uploadID"),Aggregates.sort(Sorts.ascending("_id"))
                ));
        List<String> lst = new LinkedList<>();
        for(Document d: documents) {
            lst.add(d.getString("_id"));
        }
        return lst;
//        return JSON.serialize(plantCollection.distinct("uploadID","".getClass()));
    }

public List<SyncNodeDetails> getNodeDetails(String lifeCycle) {
    UnwindOptions options = new UnwindOptions();
    options.preserveNullAndEmptyArrays(true);
    Document group = new Document("$group",new Document(SyncAttrs.ID,new Document("_id","$_id").append("host","$host").append("node","$node").append("state","$state")
                    .append("concurrencyLevel","$concurrencyLevel").append("totalHeapSize","$totalHeapSize")
                    .append("usedHeapSize","$usedHeapSize").append("lifeCycle","$lifeCycle"))
                    .append("eventArr",new Document("$addToSet","$event_docs")));
    return migrationNodeMapping.aggregate(Arrays.asList(Aggregates.match(Filters.eq(SyncAttrs.LIFE_CYCLE,lifeCycle)),Aggregates.unwind("$activeEvents",options),Aggregates.lookup("SyncEvents","activeEvents","_id","event_docs"),Aggregates.unwind("$event_docs",group,Aggregates.project(new Document("node","$_id").append("events","$eventArr").append("_id",false))),SyncNodeDetails.class)
            .into(new ArrayList<SyncNodeDetails>());
}

public Syncmarker getEventStats(ObjectId eventId) {
    Document group = new Document("$group",null).append(SyncAttrs.TOTAL_ROWS,new Document("$sum","$marker.totalRows"))
                    .append(SyncAttrs.ROWS_READ,"$marker.rowsRead"))
                    .append(SyncAttrs.ROWS_DUMPED,"$marker.rowsDumped"))
                    .append(SyncAttrs.START_TIME,new Document("$min","$marker.startTime"))
                    .append(SyncAttrs.END_TIME,new Document("$max","$marker.endTime")));
    return syncEvents.aggregate(Arrays.asList(Aggregates.match(Filters.eq(SyncAttrs.PARENT_EVENT_ID,eventId)),Aggregates.project(Projections.include(SyncAttrs.MARKER)),group),Syncmarker.class).first();
}

public List<SyncError> getEventErrors(ObjectId eventId) {
    Document group = new Document("$group",null).append(SyncAttrs.ERRORS,"$errors")));
    return syncEvents.aggregate(
            Arrays.asList(Aggregates.match(Filters.eq(SyncAttrs.PARENT_EVENT_ID,Aggregates.unwind("$errors"),Aggregates
                            .project(Projections.include(SyncAttrs.ERRORS)),Aggregates.project(new Document(SyncAttrs.ERROR_MESSAGE,"$errors.errorMessage")
                            .append(SyncAttrs.TRACE,"$errors.trace")
                            .append(SyncAttrs.THREAD_NAME,"$errors.threadName"))),SyncError.class).allowdiskUse(true).into(new ArrayList<SyncError>());
}

public String getAverageAgeByCompany() {
    AggregateIterable<Document> documents
            = userCollection.aggregate(
            Arrays.asList(
                    Aggregates.group("$company",Aggregates.sort(Sorts.ascending("_id"))
            ));
    System.err.println(JSON.serialize(documents));
    return JSON.serialize(documents);
}

@Test
public void testAggregate()
{
    List<Document> docList = new ArrayList<Document>();
    coll.aggregate(Arrays.asList(Aggregates.match(Filters.eq("name","Alto")),Aggregates.group("color",Accumulators.sum("count",1)))).into(docList);

    assertEquals(1,docList.size());

    docList.clear();

    Document first = coll
        .aggregate(Arrays.asList(Aggregates.match(Filters.eq("name",1))),Document.class)
        .allowdiskUse(true).batchSize(12).bypassDocumentValidation(true).collation(Collation.builder().build())
        .first();
    Assert.assertNotNull(first);

    first = coll
        .aggregate(Arrays.asList(Aggregates.match(Filters.eq("name",Document.class)
        .allowdiskUse(true).batchSize(12).bypassDocumentValidation(true).collation(Collation.builder().build())
        .map(new Function<Document,Document>()
        {
            @Override
            public Document apply(Document t)
            {
                t.put("hello","world");
                return t;
            }
        }).first();
    Assert.assertNotNull(first);

}

@Test
public void testFilterParametersBsonDocumentMatchInAndEq()
{

    Bson in = Filters.in("dateKey","593898622313868b72a296ad","593898622313868b72a296b4");
    Bson eq = Filters.eq("eqfield","123");
    Bson and = Filters.and(in,eq);

    Bson match = Aggregates.match(and);
    BsonDocument filterParameters = MongoUtilities.filterParameters(match);

    assertEquals("{ \"$match\" : { \"dateKey\" : { \"$in\" : [\"*?\"] },\"eqfield\" : \"?\" } }",filterParameters.toString());

}

@Test
public void testFilterParametersBsonDocumentMatchInorEq()
{

    Bson in = Filters.in("dateKey","123");
    Bson or = Filters.or(in,eq);

    Bson match = Aggregates.match(or);
    BsonDocument filterParameters = MongoUtilities.filterParameters(match);

    assertEquals("{ \"$match\" : { \"$or\" : [{ \"dateKey\" : { \"$in\" : [\"*?\"] } },{ \"eqfield\" : \"?\" }] } }",filterParameters.toString());

}

private PaginatedCollection<Application> findApplications(Bson filter) {

    final List<Bson> pipeline = new ArrayList<>(2);
    if (filter != ALL) {
      pipeline.add(Aggregates.match(filter));
    }
    pipeline.add(Aggregates.lookup(Collections.GROUPS,"group","group"));

    return RestHelper.paginatedList(database
        .getCollection(Collections.APPLICATIONS)
        .aggregate(pipeline)
        .map(Application::new)
    );
  }

private Document findApplication(Bson filter) {
  final Document bson = database.getCollection(Collections.APPLICATIONS)
     .aggregate(Lists.newArrayList(
         Aggregates.match(filter),Aggregates.lookup(Collections.GROUPS,"group")
     )).first();
  if (bson == null) {
    throw new WebApplicationException(Status.NOT_FOUND);
  }
  return bson;
}

private Map<String,Group> getAllGroups() {
  return Maps.uniqueIndex(database
    .getCollection(Collections.GROUPS)
    .aggregate(Lists.newArrayList(
      Aggregates.lookup(Collections.APPLICATIONS,"applications"),Aggregates.lookup(Collections.ASSETS,"assets")
    )).map(Group::new),t->t.id
  );
}

private PaginatedCollection<Tag> getTags() {
  return RestHelper.paginatedList(database
    .getCollection(Collections.GROUPS)
    .aggregate(Lists.newArrayList(
      Aggregates.unwind("$tags"),Aggregates.group("$tags")
    )).map(t->new Tag(t.getString("_id")))
  );
}

private List<String> getRootGroupIds(Optional<Function<String,Bson>> filterProvider) {

    final List<Bson> pipeline = new ArrayList<>(5);

    /*
     * Optional filter,needs to be applied both before and after self join to include
     * groups with inbound links from non tagged groups
     */
    Bson inboundLinksFilter = Filters.size("inbound_links",0);
    if (filterProvider.isPresent()) {

      final Bson tagFilter = filterProvider.get().apply("tags");
      pipeline.add(Aggregates.match(tagFilter));

      final Bson inboundLinksTagFilter = filterProvider.get().apply("inbound_links.tags");
      inboundLinksFilter = Filters.or(inboundLinksFilter,Filters.not(inboundLinksTagFilter));
    }

    // Unwind groups field to be able to self-join
    pipeline.add(Aggregates.unwind("$groups",new UnwindOptions().preserveNullAndEmptyArrays(true)));

    // Self join on inbound references: group.groups -> group.id and filter no inbound references
    pipeline.add(Aggregates.lookup(Collections.GROUPS,"groups.id","inbound_links"));
    pipeline.add(Aggregates.match(inboundLinksFilter));

    // Group on id to get distinct group names
    pipeline.add(Aggregates.group("$id"));

    return database
      .getCollection(Collections.GROUPS)
      .aggregate(pipeline)
      .map(t->t.getString("_id"))
      .into(Lists.newArrayList());
  }

private PaginatedCollection<String> getEnvironments() {
  return RestHelper.paginatedList(database
  .getCollection(Collections.SERVERS)
  .aggregate(Lists.newArrayList(
      Aggregates.group("$environment")
    )).map(t->t.getString("_id"))
  );
}

private PaginatedCollection<Asset> findAssets(Bson filter) {

    final List<Bson> pipeline = new ArrayList<>(2);
    if (filter != ALL) {
      pipeline.add(Aggregates.match(filter));
    }
    pipeline.add(Aggregates.lookup(Collections.GROUPS,"group"));

    return RestHelper.paginatedList(database
        .getCollection(Collections.ASSETS)
        .aggregate(pipeline)
        .map(Asset::new)
    );
  }

private Document findAsset(Bson filter) {
  final Document bson = database.getCollection(Collections.ASSETS)
      .aggregate(Lists.newArrayList(
          Aggregates.match(filter),"group")
      )).first();
  if (bson == null) {
    throw new WebApplicationException(Status.NOT_FOUND);
  }
  return bson;
}

/**
 * Create a pipeline query node based on a StatementPattern.
 * @param collection The collection of triples to query.
 * @param baseSP The leaf node in the query tree.
 */
public AggregationPipelineQueryNode(MongoCollection<Document> collection,StatementPattern baseSP) {
    this.collection = Preconditions.checkNotNull(collection);
    Preconditions.checkNotNull(baseSP);
    this.varToOriginalName = HashBiMap.create();
    StatementvarMapping mapping = new StatementvarMapping(baseSP,varToOriginalName);
    this.assuredBindingNames = new HashSet<>(mapping.varNames());
    this.bindingNames = new HashSet<>(mapping.varNames());
    this.pipeline = new LinkedList<>();
    this.pipeline.add(Aggregates.match(getMatchExpression(baseSP)));
    this.pipeline.add(Aggregates.project(mapping.getProjectExpression()));
}

/**
 * Add a $group step to filter out redundant solutions.
 * @return True if the distinct operation was successfully appended.
 */
public boolean distinct() {
    List<String> key = new LinkedList<>();
    for (String varName : bindingNames) {
        key.add(hashFieldExpr(varName));
    }
    List<BsonField> reduceOps = new LinkedList<>();
    for (String field : FIELDS) {
        reduceOps.add(new BsonField(field,"$" + field)));
    }
    pipeline.add(Aggregates.group(new Document("$concat",key),reduceOps));
    return true;
}

/**
 * Add a join with an individual {@link StatementPattern} to the pipeline.
 * @param sp The statement pattern to join with
 * @return true if the join was successfully added to the pipeline.
 */
public boolean joinWith(StatementPattern sp) {
    Preconditions.checkNotNull(sp);
    // 1. Determine shared variables and new variables
    StatementvarMapping spMap = new StatementvarMapping(sp,varToOriginalName);
    NavigableSet<String> sharedVars = new ConcurrentSkipListSet<>(spMap.varNames());
    sharedVars.retainAll(assuredBindingNames);
    // 2. Join on one shared variable
    String joinKey =  sharedVars.pollFirst();
    String collectionName = collection.getNamespace().getCollectionName();
    Bson join;
    if (joinKey == null) {
        return false;
    }
    else {
        join = Aggregates.lookup(collectionName,HASHES + "." + joinKey,spMap.hashField(joinKey),JOINED_TRIPLE);
    }
    pipeline.add(join);
    // 3. Unwind the joined triples so each document represents a binding
    //   set (solution) from the base branch and a triple that may match.
    pipeline.add(Aggregates.unwind("$" + JOINED_TRIPLE));
    // 4. (Optional) If there are any shared variables that weren't used as
    //   the join key,project all existing fields plus a new field that
    //   tests the equality of those shared variables.
    BasicDBObject matchOpts = getMatchExpression(sp,JOINED_TRIPLE);
    if (!sharedVars.isEmpty()) {
        List<Bson> eqTests = new LinkedList<>();
        for (String varName : sharedVars) {
            String oldField = valueFieldExpr(varName);
            String newField = joinFieldExpr(spMap.valueField(varName));
            Bson eqTest = new Document("$eq",Arrays.asList(oldField,newField));
            eqTests.add(eqTest);
        }
        Bson eqProjectOpts = Projections.fields(
                Projections.computed(FIELDS_MATCH,Filters.and(eqTests)),Projections.include(JOINED_TRIPLE,VALUES,HASHES,TYPES,LEVEL,TIMESTAMP));
        pipeline.add(Aggregates.project(eqProjectOpts));
        matchOpts.put(FIELDS_MATCH,true);
    }
    // 5. Filter for solutions whose triples match the joined statement
    //  pattern,and,if applicable,whose additional shared variables
    //  match the current solution.
    pipeline.add(Aggregates.match(matchOpts));
    // 6. Project the results to include variables from the new SP (with
    // appropriate renaming) and variables referenced only in the base
    // pipeline (with prevIoUs names).
    Bson finalProjectOpts = new StatementvarMapping(sp,varToOriginalName)
            .getProjectExpression(assuredBindingNames,str -> joinFieldExpr(str));
    assuredBindingNames.addAll(spMap.varNames());
    bindingNames.addAll(spMap.varNames());
    pipeline.add(Aggregates.project(finalProjectOpts));
    return true;
}

/**
 * Add a SPARQL filter to the pipeline,if possible. A filter eliminates
 * results that don't satisfy a given condition. Not all conditional
 * expressions are supported. If unsupported expressions are used in the
 * filter,the pipeline will remain unchanged and this method will return
 * false. Currently only supports binary {@link Compare} conditions among
 * variables and/or literals.
 * @param condition The filter condition
 * @return True if the filter was successfully converted into a pipeline
 *  step,false otherwise.
 */
public boolean filter(ValueExpr condition) {
    if (condition instanceof Compare) {
        Compare compare = (Compare) condition;
        Compare.CompareOp operator = compare.getoperator();
        Object leftArg = valueFieldExpr(compare.getLeftArg());
        Object rightArg = valueFieldExpr(compare.getRightArg());
        if (leftArg == null || rightArg == null) {
            // unsupported value expression,can't convert filter
            return false;
        }
        final String opFunc;
        switch (operator) {
        case EQ:
            opFunc = "$eq";
            break;
        case NE:
            opFunc = "$ne";
            break;
        case LT:
            opFunc = "$lt";
            break;
        case LE:
            opFunc = "$le";
            break;
        case GT:
            opFunc = "$gt";
            break;
        case GE:
            opFunc = "$ge";
            break;
        default:
            // unrecognized comparison operator,can't convert filter
            return false;
        }
        Document compareDoc = new Document(opFunc,Arrays.asList(leftArg,rightArg));
        pipeline.add(Aggregates.project(Projections.fields(
                Projections.computed("FILTER",compareDoc),Projections.include(VALUES,TIMESTAMP))));
        pipeline.add(Aggregates.match(new Document("FILTER",true)));
        pipeline.add(Aggregates.project(Projections.fields(
                Projections.include(VALUES,TIMESTAMP))));
        return true;
    }
    return false;
}

/**
 * Given that the current state of the pipeline produces data that can be
 * interpreted as triples,add a project step to map each result from the
 * intermediate result structure to a structure that can be stored in the
 * triple store. Does not modify the internal pipeline,which will still
 * produce intermediate results suitable for query evaluation.
 * @param timestamp Attach this timestamp to the resulting triples.
 * @param requireNew If true,add an additional step to check constructed
 *  triples against existing triples and only include new ones in the
 *  result. Adds a potentially expensive $lookup step.
 * @throws IllegalStateException if the results produced by the current
 *  pipeline do not have variable names allowing them to be interpreted as
 *  triples (i.e. "subject","predicate",and "object").
 */
public List<Bson> getTriplePipeline(long timestamp,boolean requireNew) {
    if (!assuredBindingNames.contains(SUBJECT)
            || !assuredBindingNames.contains(PREDICATE)
            || !assuredBindingNames.contains(OBJECT)) {
        throw new IllegalStateException("Current pipeline does not produce "
                + "records that can be converted into triples.\n"
                + "required variable names: <" + SUBJECT + "," + PREDICATE
                + "," + OBJECT + ">\nCurrent variable names: "
                + assuredBindingNames);
    }
    List<Bson> triplePipeline = new LinkedList<>(pipeline);
    List<Bson> fields = new LinkedList<>();
    fields.add(Projections.computed(SUBJECT,valueFieldExpr(SUBJECT)));
    fields.add(Projections.computed(SUBJECT_HASH,hashFieldExpr(SUBJECT)));
    fields.add(Projections.computed(PREDICATE,valueFieldExpr(PREDICATE)));
    fields.add(Projections.computed(PREDICATE_HASH,hashFieldExpr(PREDICATE)));
    fields.add(Projections.computed(OBJECT,valueFieldExpr(OBJECT)));
    fields.add(Projections.computed(OBJECT_HASH,hashFieldExpr(OBJECT)));
    fields.add(Projections.computed(OBJECT_TYPE,ConditionalOperators.ifNull(typeFieldExpr(OBJECT),DEFAULT_TYPE)));
    fields.add(Projections.computed(CONTEXT,DEFAULT_CONTEXT));
    fields.add(Projections.computed(STATEMENT_MetaDATA,DEFAULT_MetaDATA));
    fields.add(DEFAULT_DV);
    fields.add(Projections.computed(TIMESTAMP,new Document("$literal",timestamp)));
    fields.add(Projections.computed(LEVEL,new Document("$add",Arrays.asList("$" + LEVEL,1))));
    triplePipeline.add(Aggregates.project(Projections.fields(fields)));
    if (requireNew) {
        // Prune any triples that already exist in the data store
        String collectionName = collection.getNamespace().getCollectionName();
        Bson includeAll = Projections.include(SUBJECT,SUBJECT_HASH,PREDICATE,PREDICATE_HASH,OBJECT,OBJECT_HASH,OBJECT_TYPE,CONTEXT,STATEMENT_MetaDATA,DOCUMENT_VISIBILITY,TIMESTAMP,LEVEL);
        List<Bson> eqTests = new LinkedList<>();
        eqTests.add(new Document("$eq",Arrays.asList("$$this." + PREDICATE_HASH,"$" + PREDICATE_HASH)));
        eqTests.add(new Document("$eq",Arrays.asList("$$this." + OBJECT_HASH,"$" + OBJECT_HASH)));
        Bson redundantFilter = new Document("$filter",new Document("input","$" + JOINED_TRIPLE)
                .append("as","this").append("cond",new Document("$and",eqTests)));
        triplePipeline.add(Aggregates.lookup(collectionName,JOINED_TRIPLE));
        String numRedundant = "REDUNDANT";
        triplePipeline.add(Aggregates.project(Projections.fields(includeAll,Projections.computed(numRedundant,new Document("$size",redundantFilter)))));
        triplePipeline.add(Aggregates.match(Filters.eq(numRedundant,0)));
        triplePipeline.add(Aggregates.project(Projections.fields(includeAll)));
    }
    return triplePipeline;
}

@Test
public void testFilterParametersBsonDocumentMatchIn()
{

    Bson match = Aggregates.match(Filters.in("dateKey","593898622313868b72a296b4"));
    BsonDocument filterParameters = MongoUtilities.filterParameters(match);

    assertEquals("{ \"$match\" : { \"dateKey\" : { \"$in\" : [\"*?\"] } } }",filterParameters.toString());

}

@Test
public void testFilterParametersBsonDocumentGroup()
{

    Bson group = Aggregates.group("_id",Accumulators.sum("totalQuantity","$quantity"));
    BsonDocument filterParameters = MongoUtilities.filterParameters(group);

    assertEquals("{ \"$group\" : { \"_id\" : \"_id\",\"totalQuantity\" : { \"$sum\" : \"$quantity\" } } }",filterParameters.toString());

}

@Test
public void testFilterParametersBsonDocumentLookup()
{      

    Bson lookup = Aggregates.lookup("fromField","localField","foreignField","as");
    BsonDocument filterParameters = MongoUtilities.filterParameters(lookup);

    assertEquals("{ \"$lookup\" : { \"from\" : \"fromField\",\"localField\" : \"localField\",\"foreignField\" : \"foreignField\",\"as\" : \"as\" } }",filterParameters.toString());

}

/**
 * Add a step to the end of the current pipeline which prunes the results
 * according to the recorded derivation level of their sources. At least one
 * triple that was used to construct the result must have a derivation level
 * at least as high as the parameter,indicating that it was derived via
 * that many steps from the original data. (A value of zero is equivalent to
 * input data that was not derived at all.) Use in conjunction with
 * getTriplePipeline (which sets source level for generated triples) to
 * avoid repeatedly deriving the same results.
 * @param requiredLevel required derivation depth. Reject a solution to the
 *  query if all of the triples involved in producing that solution have a
 *  lower derivation depth than this. If zero,does nothing.
 */
public void requireSourceDerivationDepth(int requiredLevel) {
    if (requiredLevel > 0) {
        pipeline.add(Aggregates.match(new Document(LEVEL,new Document("$gte",requiredLevel))));
    }
}

/**
 * Add a step to the end of the current pipeline which prunes the results
 * according to the timestamps of their sources. At least one triple that
 * was used to construct the result must have a timestamp at least as
 * recent as the parameter. Use in iterative applications to avoid deriving
 * solutions that would have been generated in an earlier iteration.
 * @param t Minimum required timestamp. Reject a solution to the query if
 *  all of the triples involved in producing that solution have an earlier
 *  timestamp than this.
 */
public void requireSourceTimestamp(long t) {
    pipeline.add(Aggregates.match(new Document(TIMESTAMP,t))));
}