JAVA-5990 Hybrid Search Score Fusion

driver-core/src/main/com/mongodb/client/model/Aggregates.java

@@ -46,9 +46,11 @@
 import org.bson.codecs.configuration.CodecRegistry;
 import org.bson.conversions.Bson;
 
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 
 import static com.mongodb.assertions.Assertions.assertTrue;
 import static com.mongodb.assertions.Assertions.isTrueArgument;
@@ -1093,6 +1095,55 @@ public static Bson rerank(
         return new RerankBson(query, paths, numDocsToRerank, model);
     }
 
+    /**
+     * Creates a {@code $scoreFusion} pipeline stage, which combines the results of the given input pipelines,
+     * normalizing and combining the scores they produce. It must be the first stage of an aggregation pipeline
+     * run on a collection, and the input pipelines must be scored selection pipelines on the same collection,
+     * e.g., starting with a {@link #search(SearchOperator, SearchOptions) $search}
+     * or {@link #vectorSearch(FieldSearchPath, Iterable, String, long, VectorSearchOptions) $vectorSearch} stage.
+     * You may use the {@code $meta: "score"} expression to extract the combined score of a document.
+     *
+     * @param pipelines The non-empty input pipelines with unique names.
+     * @param normalization The way in which the scores produced by the input pipelines are normalized.
+     * @return The {@code $scoreFusion} pipeline stage.
+     * @mongodb.driver.manual reference/operator/aggregation/scoreFusion/ $scoreFusion
+     * @mongodb.server.release 8.2
+     * @since 5.10
+     */
+    public static Bson scoreFusion(final List<FusionPipeline> pipelines, final ScoreNormalization normalization) {
+        return scoreFusion(pipelines, normalization, ScoreFusionOptions.scoreFusionOptions());
+    }
+
+    /**
+     * Creates a {@code $scoreFusion} pipeline stage, which combines the results of the given input pipelines,
+     * normalizing and combining the scores they produce. It must be the first stage of an aggregation pipeline
+     * run on a collection, and the input pipelines must be scored selection pipelines on the same collection,
+     * e.g., starting with a {@link #search(SearchOperator, SearchOptions) $search}
+     * or {@link #vectorSearch(FieldSearchPath, Iterable, String, long, VectorSearchOptions) $vectorSearch} stage.
+     * You may use the {@code $meta: "score"} expression to extract the combined score of a document.
+     *
+     * @param pipelines The non-empty input pipelines with unique names.
+     * @param normalization The way in which the scores produced by the input pipelines are normalized.
+     * @param options Optional {@code $scoreFusion} pipeline stage fields.
+     * @return The {@code $scoreFusion} pipeline stage.
+     * @mongodb.driver.manual reference/operator/aggregation/scoreFusion/ $scoreFusion
+     * @mongodb.server.release 8.2
+     * @since 5.10
+     */
+    public static Bson scoreFusion(final List<FusionPipeline> pipelines, final ScoreNormalization normalization,
+            final ScoreFusionOptions options) {
+        notNull("pipelines", pipelines);
+        isTrueArgument("pipelines must not be empty", !pipelines.isEmpty());
+        Set<String> names = new HashSet<>();
+        for (FusionPipeline pipeline : pipelines) {
+            notNull("pipeline", pipeline);
+            isTrueArgument("pipeline names must be unique", names.add(pipeline.getName()));
+        }
+        notNull("normalization", normalization);
+        notNull("options", options);
+        return new ScoreFusionStage(pipelines, normalization, options);
+    }
+
     /**
      * Creates an $unset pipeline stage that removes/excludes fields from documents
      *
@@ -2344,6 +2395,45 @@ public String toString() {
         }
     }
 
+    private static final class ScoreFusionStage implements Bson {
+        private final List<FusionPipeline> pipelines;
+        private final ScoreNormalization normalization;
+        private final ScoreFusionOptions options;
+
+        ScoreFusionStage(final List<FusionPipeline> pipelines, final ScoreNormalization normalization,
+                final ScoreFusionOptions options) {
+            this.pipelines = pipelines;
+            this.normalization = normalization;
+            this.options = options;
+        }
+
+        @Override
+        public <TDocument> BsonDocument toBsonDocument(final Class<TDocument> documentClass, final CodecRegistry codecRegistry) {
+            BsonDocument pipelinesDoc = new BsonDocument();
+            for (FusionPipeline pipeline : pipelines) {
+                BsonArray stages = new BsonArray();
+                for (Bson stage : pipeline.getPipeline()) {
+                    stages.add(stage.toBsonDocument(documentClass, codecRegistry));
+                }
+                pipelinesDoc.append(pipeline.getName(), stages);
+            }
+            BsonDocument specificationDoc = new BsonDocument("input",
+                    new BsonDocument("pipelines", pipelinesDoc)
+                            .append("normalization", new BsonString(normalization.getValue())));
+            specificationDoc.putAll(options.toBsonDocument(documentClass, codecRegistry));
+            return new BsonDocument("$scoreFusion", specificationDoc);
+        }
+
+        @Override
+        public String toString() {
+            return "Stage{name=$scoreFusion"
+                    + ", pipelines=" + pipelines
+                    + ", normalization=" + normalization
+                    + ", options=" + options
+                    + '}';
+        }
+    }
+
     private static class RerankBson implements Bson {
         private final RerankQuery query;
         private final List<String> paths;

driver-core/src/main/com/mongodb/client/model/FusionPipeline.java

@@ -0,0 +1,123 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.mongodb.client.model;
+
+import org.bson.conversions.Bson;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+import static com.mongodb.assertions.Assertions.isTrueArgument;
+import static com.mongodb.assertions.Assertions.notNull;
+import static java.util.Arrays.asList;
+import static java.util.Collections.unmodifiableList;
+
+/**
+ * A named aggregation pipeline used as an input to a fusion pipeline stage, e.g.,
+ * {@link Aggregates#scoreFusion(List, ScoreNormalization, ScoreFusionOptions) $scoreFusion}.
+ * The name uniquely identifies the pipeline within the stage and may be referred to
+ * by other parts of the stage, e.g., as the {@code "$$name"} variable in a
+ * {@linkplain ScoreFusionCombination#expression(Bson) combination expression}.
+ * A name must not be empty, must not start with {@code $}, and must not contain {@code .}
+ * or the null character.
+ *
+ * @since 5.10
+ * @mongodb.server.release 8.2
+ */
+public final class FusionPipeline {
+    private final String name;
+    private final List<Bson> pipeline;
+
+    /**
+     * Creates a new {@link FusionPipeline}.
+     *
+     * @param name The pipeline name, unique within the containing stage. It must not be empty,
+     * must not start with {@code $}, and must not contain {@code .} or the null character.
+     * @param pipeline The non-empty pipeline.
+     * @return The requested {@link FusionPipeline}.
+     */
+    public static FusionPipeline of(final String name, final List<? extends Bson> pipeline) {
+        return new FusionPipeline(name, pipeline);
+    }
+
+    /**
+     * Creates a new {@link FusionPipeline}.
+     *
+     * @param name The pipeline name, unique within the containing stage. It must not be empty,
+     * must not start with {@code $}, and must not contain {@code .} or the null character.
+     * @param pipeline The non-empty pipeline.
+     * @return The requested {@link FusionPipeline}.
+     */
+    public static FusionPipeline of(final String name, final Bson... pipeline) {
+        return new FusionPipeline(name, asList(pipeline));
+    }
+
+    private FusionPipeline(final String name, final List<? extends Bson> pipeline) {
+        notNull("name", name);
+        isTrueArgument("name must not be empty", !name.trim().isEmpty());
+        isTrueArgument("name must not start with '$'", !name.startsWith("$"));
+        isTrueArgument("name must not contain '.'", !name.contains("."));
+        isTrueArgument("name must not contain the null character", name.indexOf('\u0000') < 0);
+        notNull("pipeline", pipeline);
+        isTrueArgument("pipeline must not be empty", !pipeline.isEmpty());
+        for (Bson stage : pipeline) {
+            notNull("stage", stage);
+        }
+        this.name = name;
+        this.pipeline = unmodifiableList(new ArrayList<Bson>(pipeline));
+    }
+
+    /**
+     * @return the pipeline name
+     */
+    public String getName() {
+        return name;
+    }
+
+    /**
+     * @return the pipeline
+     */
+    public List<? extends Bson> getPipeline() {
+        return pipeline;
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        FusionPipeline that = (FusionPipeline) o;
+        return name.equals(that.name) && pipeline.equals(that.pipeline);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(name, pipeline);
+    }
+
+    @Override
+    public String toString() {
+        return "FusionPipeline{"
+                + "name='" + name + '\''
+                + ", pipeline=" + pipeline
+                + '}';
+    }
+}

driver-core/src/main/com/mongodb/client/model/ScoreFusionCombination.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.mongodb.client.model;
+
+import com.mongodb.annotations.Sealed;
+import org.bson.BsonString;
+import org.bson.Document;
+import org.bson.conversions.Bson;
+
+import static com.mongodb.assertions.Assertions.notNull;
+
+/**
+ * The way in which the normalized scores produced by the input pipelines of the
+ * {@linkplain Aggregates#scoreFusion(java.util.List, ScoreNormalization, ScoreFusionOptions) $scoreFusion}
+ * stage are combined into the final score. The server rejects specifying both
+ * {@linkplain #weighted(Bson) weights} and an {@linkplain #expression(Bson) expression},
+ * which is why they are separate factory methods.
+ *
+ * @see ScoreFusionOptions#combination(ScoreFusionCombination)
+ * @since 5.10
+ * @mongodb.server.release 8.2
+ */
+@Sealed
+public interface ScoreFusionCombination extends Bson {
+    /**
+     * Returns a {@link WeightedScoreFusionCombination} combining the scores using per-pipeline weights.
+     *
+     * @param weights A document mapping {@linkplain FusionPipeline#getName() pipeline names} to non-negative
+     * numeric weights. Pipelines not mentioned have the server-default weight 1.
+     * @return The requested {@link WeightedScoreFusionCombination}.
+     */
+    static WeightedScoreFusionCombination weighted(final Bson weights) {
+        return new ScoreFusionConstructibleBson(new Document("weights", notNull("weights", weights)));
+    }
+
+    /**
+     * Returns a {@link ScoreFusionCombination} combining the scores using a custom expression.
+     * The normalized, weighted score of each input pipeline is available to the expression
+     * as the variable named after the pipeline, e.g., {@code "$$name"}.
+     *
+     * @param expression The combination expression.
+     * @return The requested {@link ScoreFusionCombination}.
+     */
+    static ScoreFusionCombination expression(final Bson expression) {
+        return new ScoreFusionConstructibleBson(new Document("method", new BsonString("expression"))
+                .append("expression", notNull("expression", expression)));
+    }
+}

driver-core/src/main/com/mongodb/client/model/ScoreFusionConstructibleBson.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.mongodb.client.model;
+
+import com.mongodb.annotations.Immutable;
+import com.mongodb.internal.client.model.AbstractConstructibleBson;
+import org.bson.BsonBoolean;
+import org.bson.BsonDocument;
+import org.bson.BsonString;
+import org.bson.Document;
+import org.bson.conversions.Bson;
+
+import static com.mongodb.assertions.Assertions.notNull;
+
+final class ScoreFusionConstructibleBson extends AbstractConstructibleBson<ScoreFusionConstructibleBson>
+        implements ScoreFusionOptions, WeightedScoreFusionCombination {
+    /**
+     * An {@linkplain Immutable immutable} {@link BsonDocument#isEmpty() empty} instance.
+     */
+    static final ScoreFusionConstructibleBson EMPTY_IMMUTABLE =
+            new ScoreFusionConstructibleBson(AbstractConstructibleBson.EMPTY_IMMUTABLE);
+
+    ScoreFusionConstructibleBson(final Bson base) {
+        super(base);
+    }
+
+    private ScoreFusionConstructibleBson(final Bson base, final Document appended) {
+        super(base, appended);
+    }
+
+    @Override
+    protected ScoreFusionConstructibleBson newSelf(final Bson base, final Document appended) {
+        return new ScoreFusionConstructibleBson(base, appended);
+    }
+
+    @Override
+    public ScoreFusionOptions combination(final ScoreFusionCombination combination) {
+        return newAppended("combination", notNull("combination", combination));
+    }
+
+    @Override
+    public ScoreFusionOptions scoreDetails(final boolean scoreDetails) {
+        return newAppended("scoreDetails", BsonBoolean.valueOf(scoreDetails));
+    }
+
+    @Override
+    public ScoreFusionOptions option(final String name, final Object value) {
+        return newAppended(notNull("name", name), notNull("value", value));
+    }
+
+    @Override
+    public WeightedScoreFusionCombination avg() {
+        return newAppended("method", new BsonString("avg"));
+    }
+}