Native Aggregation

Describes how to execute the Bson aggregation stage of the MongoDB Driver from the first stage without reimplementing new Driver features in DSL.

Driver-native root aggregation

aggregation() executes the Bson stages returned by the MongoDB Driver in the order they are called from the first stage. The DSL does not reimplement Driver aggregation operators but handles pipeline execution, mapping, Reactor publisher, and preview/explain connections. Document can also be passed directly as it is Bson.

java

Flux<User> result = dsl
    .executeEntity(User.class, MongoTemplateName.FRONT)
    .aggregation()
    .stage(Aggregates.match(Filters.eq("status", "ACTIVE")))
    .stage(Aggregates.sort(Sorts.descending("createdAt")))
    .stage(Aggregates.limit(20))
    .execute();

Result mapping

execute()

Maps the result to the source entity type.

execute(ResultType.class)

Maps to a separate result type when the shape of the aggregation result differs from the source entity.

executeDocument()

Returns the raw MongoDB Document as is.

$score / $scoreFusion of MongoDB Driver 5.10

1.0.0 does not create a dedicated wrapper for $score and $scoreFusion. Since Driver 5.10's Aggregates.score(...) / Aggregates.scoreFusion(...) already returns a Bson stage, it is passed directly to stage(...).

java

.aggregation()
.stage(Aggregates.score("$rating"))
.executeDocument();

java

Bson searchStage = Aggregates.search(
    SearchOperator.text(SearchPath.fieldPath("title"), "mongodb"),
    SearchOptions.searchOptions().index("search-index")
);

Bson vectorStage = Aggregates.vectorSearch(
    SearchPath.fieldPath("embedding"),
    List.of(0.1D, 0.2D, 0.3D),
    "vector-index",
    10L,
    VectorSearchOptions.exactVectorSearchOptions()
);

Flux<Document> result = dsl
    .executeEntity(User.class, MongoTemplateName.FRONT)
    .aggregation()
    .stage(
        Aggregates.scoreFusion(
            List.of(
                FusionPipeline.of("text", searchStage),
                FusionPipeline.of("vector", vectorStage)
            ),
            ScoreNormalization.SIGMOID
        )
    )
    .executeDocument();

Driver API and server support versions are separate.

$score / $scoreFusion are features of MongoDB 8.2+ server. Just because the Driver 5.10 API is compiled does not mean it can run on earlier servers.

Add Bson stage after Search / Vector

java

.search("search-index")
.text(text -> text.path("title").query("mongodb"))
.stage(Aggregates.score("$rating"))
.findAll()
.execute();

.vectorSearch("vector-index")
.path("embedding")
.queryVector(vector)
.limit(20)
.exact()
.stage(Aggregates.score("$rating"))
.findAll()
.execute();

The stage(Bson) of Search/Vector comes immediately after the mandatory first stage $search/$vectorSearch. Features that require the stage itself to be the first in the pipeline use aggregation().

Aggregation publisher options

java

.aggregation()
.readPreference(ReadPreference.secondaryPreferred())
.isAllowDiskUse(true)
.customizeAggregation(publisher -> publisher.batchSize(500))

preview() and actual MongoDB explain() are also supported in the root aggregation builder.