Native Aggregation

MongoDB Driver의 Bson aggregation stage를 첫 단계부터 그대로 실행하고 신규 Driver 기능을 DSL 재구현 없이 사용하는 방법을 설명합니다.

Driver-native root aggregation

aggregation()은 MongoDB Driver가 반환하는 Bson stage를 첫 stage부터 호출 순서 그대로 실행합니다. DSL이 Driver aggregation operator를 다시 구현하지 않고 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();

결과 mapping

execute()

source entity type으로 결과를 mapping합니다.

execute(ResultType.class)

aggregation 결과 shape가 source entity와 다를 때 별도 result type으로 mapping합니다.

executeDocument()

raw MongoDB Document를 그대로 반환합니다.

MongoDB Driver 5.10의 $score / $scoreFusion

1.0.0은 $score와 $scoreFusion 전용 wrapper를 만들지 않습니다. Driver 5.10의 Aggregates.score(...) / Aggregates.scoreFusion(...)가 이미 Bson stage를 반환하므로 그대로 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가 compile된다는 사실만으로 이전 서버에서 실행 가능한 것은 아닙니다.

Search / Vector 뒤에 Bson stage 추가

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)은 mandatory first stage인 $search/$vectorSearch 직후에 들어갑니다. $scoreFusion처럼 stage 자체가 pipeline 첫 번째여야 하는 기능은 aggregation()을 사용합니다.

Aggregation publisher options

java

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

preview()와 실제 MongoDB explain()도 root aggregation builder에서 지원합니다.