MongoDB aggregation wtih distinct value in Java?

user_834239 0
Viewed 458

I am trying to get the latest entered document based upon a unique ID. Here is the collection

{
        "_id" : "535f5d074f075c37fff4cc74",
        "senderId" : 8989898,
        "receiverId" : 8686868,
        "channelId" : "909090",
        "createdAt": 1
}
{
        "_id" : "535f5d074f075c37fff4cc74",
        "senderId" : 8989898,
        "receiverId" : 8686868,
        "channelId" : "909090"
        "createdAt": 2
}
{
        "_id" : "535f5d074f075c37fff4cc74",
        "senderId" : 8989898,
        "receiverId" : 10101010,
        "channelId" : "919191"
        "createdAt": 3
}

The result will be :

{
        "_id" : "535f5d074f075c37fff4cc74",
        "senderId" : 8989898,
        "receiverId" : 8686868,
        "channelId" : "909090"
        "createdAt": 2
}
{
        "_id" : "535f5d074f075c37fff4cc74",
        "senderId" : 8989898,
        "receiverId" : 10101010,
        "channelId" : "919191"
        "createdAt": 3
}

Providing the elements with distinct channelId keyword and the latest created with in channelId.

I tried the following approach:

Aggregation aggregation = Aggregation.newAggregation(
            Aggregation.sort(Sort.Direction.DESC, Constant.CREATED_AT),
            Aggregation.group(Constant.CHANNEL_ID));

return mongoTemplate.aggregate(aggregation, "tbl_message", Messages.class)
            .getMappedResults()
            .stream()
            .distinct()
            .collect(Collectors.toList());

The above code is just providing me distinct channelId (s) in the entire collection. How can I correct the above statement as per the requirements?

spring-boot aggregation-framework mongodb-query mongodb aggregation
1 Answers

Your example documents are invalid because you have a unique key violation on field _id. For this reason, I have renamed the field to simply id.

Sample Test Data

db.collection.insertMany([
{
        "id" : "535f5d074f075c37fff4cc74",
        "senderId" : 8989898,
        "receiverId" : 8686868,
        "channelId" : "909090",
        "createdAt": 1
},
{
        "id" : "535f5d074f075c37fff4cc74",
        "senderId" : 8989898,
        "receiverId" : 8686868,
        "channelId" : "909090",
        "createdAt": 2
},
{
        "id" : "535f5d074f075c37fff4cc74",
        "senderId" : 8989898,
        "receiverId" : 10101010,
        "channelId" : "919191",
        "createdAt": 3
}])

Aside from that, I think you were pretty close, but your GROUP BY clause needs some changes. Here is what I believe to be a working version...

Code Example

package test.barry;

public class Main {
    public static void main(String[] args) {
        com.mongodb.client.MongoClient client = connectToReplicaSet();
        com.mongodb.client.MongoDatabase db = client.getDatabase("barrydb");
        com.mongodb.client.MongoCollection<org.bson.Document> collection = db.getCollection("collection");

        com.mongodb.client.AggregateIterable<org.bson.Document> iterable1 = collection.aggregate(
            java.util.Arrays.asList(
                com.mongodb.client.model.Aggregates.sort(com.mongodb.client.model.Sorts.orderBy(com.mongodb.client.model.Sorts.ascending("id"), com.mongodb.client.model.Sorts.ascending("channelId"), com.mongodb.client.model.Sorts.descending("createdAt"))),
                com.mongodb.client.model.Aggregates.group(
                    new org.bson.BsonDocument("id", new org.bson.BsonString("$id")).append("channelId", new org.bson.BsonString("$channelId")),
                        com.mongodb.client.model.Accumulators.first("senderId", "$senderId"),
                        com.mongodb.client.model.Accumulators.first("receiverId", "$receiverId"),
                        com.mongodb.client.model.Accumulators.first("createdAt", "$createdAt"),
                        com.mongodb.client.model.Accumulators.first("originalId", "$_id")
                ),
                com.mongodb.client.model.Aggregates.project(
                    com.mongodb.client.model.Projections.fields(
                        com.mongodb.client.model.Projections.computed("_id", "$originalId"),
                        com.mongodb.client.model.Projections.computed("id", "$_id.id"),
                        com.mongodb.client.model.Projections.include("senderId", "receiverId"),
                        com.mongodb.client.model.Projections.computed("channelId", "$_id.channelId"),
                        com.mongodb.client.model.Projections.include("createdAt")
                    )
                )
            )
        );

        // AUTO-CLOSABLE TRY
        try(com.mongodb.client.MongoCursor<org.bson.Document> cursor1 = iterable1.iterator())
        {
            while (cursor1.hasNext())
            {
                org.bson.Document queriedDocument1 = cursor1.next();
                System.out.println(String.format("queriedDocument1: %s", queriedDocument1));
            }
        }
    }

    private static com.mongodb.client.MongoClient connectToReplicaSet() {
        java.util.ArrayList<com.mongodb.ServerAddress> hosts = new java.util.ArrayList<com.mongodb.ServerAddress>();
        hosts.add(new com.mongodb.ServerAddress("localhost", 50011));
        hosts.add(new com.mongodb.ServerAddress("localhost", 50012));
        hosts.add(new com.mongodb.ServerAddress("localhost", 50013));

        com.mongodb.MongoCredential mongoCredential = com.mongodb.MongoCredential.createScramSha256Credential("testuser", "admin", "mysecret".toCharArray());

        com.mongodb.MongoClientSettings mongoClientSettings = com.mongodb.MongoClientSettings.builder()
            .applyToClusterSettings(clusterSettingsBuilder -> clusterSettingsBuilder.hosts(hosts).requiredReplicaSetName("replSet"))
            .credential(mongoCredential)
            .writeConcern(com.mongodb.WriteConcern.MAJORITY)
            .readConcern(com.mongodb.ReadConcern.MAJORITY)
            .readPreference(com.mongodb.ReadPreference.primary())
            .retryWrites(true)
            .build();

        com.mongodb.client.MongoClient client = com.mongodb.client.MongoClients.create(mongoClientSettings);

        return client;
    }
}

Aggregation Explained

There are 3 stages to this aggregation

  1. Sort
  2. Group
  3. Project

The $sort stage will sort by field id ascending, then by field channelId ascending, then by createdAt descending. This will kind of group the records logically together in the desired order.

The $group stage will group by id and channelId. Because the documents have already been sorted in the first stage we can simply take the first occurrence of a group and retain the other fields (via accumulator $first).

The grouping stage leaves the documents in a less-than-desirable shape. To fix the shape, use a $project to put the documents back into desired shape.

Example output

queriedDocument1: Document{{senderId=8989898, receiverId=10101010, createdAt=3, _id=616d9387719033ca154188a2, id=535f5d074f075c37fff4cc74, channelId=919191}}
queriedDocument1: Document{{senderId=8989898, receiverId=8686868, createdAt=2, _id=616d9387719033ca154188a1, id=535f5d074f075c37fff4cc74, channelId=909090}}

Conclusion

It appears there is no additional sorting requirement for the final output, so the natural order where 3 arrives before 2 is OK. If needed, an additional $sort stage could be applied for the final output.

user_4155590
Related