原生聚合

说明如何从第一阶段开始直接执行 MongoDB Driver 的 Bson 聚合阶段，并在不重新实现 DSL 功能的情况下使用新 Driver 功能。

Driver-native 根聚合

aggregation() 从 MongoDB Driver 返回的 Bson 阶段开始按调用顺序执行。DSL 不重新实现 Driver 聚合操作符，仅负责管道执行、映射、Reactor 发布者、预览/解释连接。Document 也是 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();

结果映射

execute()

将结果映射为源实体类型。

execute(ResultType.class)

当聚合结果形状与源实体不同时，映射为单独的结果类型。

executeDocument()

原始 MongoDB Document 将被直接返回。

MongoDB Driver 5.10 的 $score / $scoreFusion

1.0.0 不会创建 $score 和 $scoreFusion 的专用包装器。Driver 5.10 的 Aggregates.score(...) / Aggregates.scoreFusion(...) 已经返回 Bson 阶段，因此直接传递给 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 和服务器支持版本是独立的。

$score / $scoreFusion 是 MongoDB 8.2+ 服务器功能。Driver 5.10 API 的编译并不意味着可以在旧版服务器上运行。

在 Search / Vector 后添加 Bson 阶段

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();

Search/Vector 的 stage(Bson) 紧跟在强制性第一个阶段 $search/$vectorSearch 之后。像 $scoreFusion 这样的功能需要将阶段本身作为管道的第一个。使用 aggregation()。

聚合发布者选项

java

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

preview() 和实际 MongoDB explain() 也在根聚合构建器中得到支持。