原生聚合

說明如何從第一階段開始直接執行 MongoDB Driver 的 Bson 聚合階段，並在不重新實現 DSL 功能的情況下使用新 Driver 功能。

Driver-native 根聚合

aggregation() 會按照 MongoDB Driver 返回的 Bson 階段的調用順序從第一階段開始執行。DSL 不重新實現 Driver 聚合運算符，只負責 pipeline 執行、mapping、Reactor publisher、preview/explain 連接。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()

將結果映射為 source entity 類型。

執行(ResultType.class)

當聚合結果形狀與來源實體不一致時，映射到單獨的結果類型。

執行文件()

直接返回原始 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() 也在根聚合構建器中得到支持。