ネイティブ集約

MongoDB DriverのBson集約ステージを最初のステージからそのまま実行し、新しいDriver機能をDSL再実装なしで使用する方法を説明します。

Driver-native root aggregation

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

source entity typeで結果をマッピングします。

execute(ResultType.class)

aggregationの結果のshapeがsource entityと異なる場合、別のresult typeにマッピングします。

executeDocument()

raw 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()を使用します。

Aggregation publisher options

java

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

preview()と実際のMongoDB explain()もルートaggregationビルダーでサポートされています。