Add harmonic mean combination

src/main/java/org/opensearch/neuralsearch/processor/combination/AbstractScoreCombinationTechnique.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.neuralsearch.processor.combination;
+
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Base class for score normalization technique
+ */
+public abstract class AbstractScoreCombinationTechnique {
+    private static final String PARAM_NAME_WEIGHTS = "weights";
+
+    /**
+     * Each technique must provide collection of supported parameters
+     * @return set of supported parameter names
+     */
+    abstract Set<String> getSupportedParams();
+
+    /**
+     * Get collection of weights based on user provided config
+     * @param params map of named parameters and their values
+     * @return collection of weights
+     */
+    protected List<Float> getWeights(final Map<String, Object> params) {
+        if (Objects.isNull(params) || params.isEmpty()) {
+            return List.of();
+        }
+        // get weights, we don't need to check for instance as it's done during validation
+        return ((List<Double>) params.getOrDefault(PARAM_NAME_WEIGHTS, List.of())).stream()
+            .map(Double::floatValue)
+            .collect(Collectors.toUnmodifiableList());
+    }
+
+    /**
+     * Validate config parameters for this technique
+     * @param params map of parameters in form of name-value
+     */
+    protected void validateParams(final Map<String, Object> params) {
+        if (Objects.isNull(params) || params.isEmpty()) {
+            return;
+        }
+        // check if only supported params are passed
+        Optional<String> optionalNotSupportedParam = params.keySet()
+            .stream()
+            .filter(paramName -> !getSupportedParams().contains(paramName))
+            .findFirst();
+        if (optionalNotSupportedParam.isPresent()) {
+            throw new IllegalArgumentException(
+                String.format(
+                    Locale.ROOT,
+                    "provided parameter for combination technique is not supported. supported parameters are [%s]",
+                    getSupportedParams().stream().collect(Collectors.joining(","))
+                )
+            );
+        }
+
+        // check param types
+        if (params.keySet().stream().anyMatch(PARAM_NAME_WEIGHTS::equalsIgnoreCase)) {
+            if (!(params.get(PARAM_NAME_WEIGHTS) instanceof List)) {
+                throw new IllegalArgumentException(
+                    String.format(Locale.ROOT, "parameter [%s] must be a collection of numbers", PARAM_NAME_WEIGHTS)
+                );
+            }
+        }
+    }
+
+    /**
+     * Get weight for sub-query based on its index in the hybrid search query. Use user provided weight or 1.0 otherwise
+     * @param weights collection of weights for sub-queries
+     * @param indexOfSubQuery 0-based index of sub-query in the Hybrid Search query
+     * @return weight for sub-query, use one that is set in processor/pipeline definition or 1.0 as default
+     */
+    protected float getWeightForSubQuery(final List<Float> weights, final int indexOfSubQuery) {
+        return indexOfSubQuery < weights.size() ? weights.get(indexOfSubQuery) : 1.0f;
+    }
+}

src/main/java/org/opensearch/neuralsearch/processor/combination/ArithmeticMeanScoreCombinationTechnique.java

@@ -6,17 +6,13 @@
 package org.opensearch.neuralsearch.processor.combination;
 
 import java.util.List;
-import java.util.Locale;
 import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
 import java.util.Set;
-import java.util.stream.Collectors;
 
 /**
  * Abstracts combination of scores based on arithmetic mean method
  */
-public class ArithmeticMeanScoreCombinationTechnique implements ScoreCombinationTechnique {
+public class ArithmeticMeanScoreCombinationTechnique extends AbstractScoreCombinationTechnique implements ScoreCombinationTechnique {
 
     public static final String TECHNIQUE_NAME = "arithmetic_mean";
     public static final String PARAM_NAME_WEIGHTS = "weights";
@@ -29,16 +25,6 @@ public ArithmeticMeanScoreCombinationTechnique(final Map<String, Object> params)
         weights = getWeights(params);
     }
 
-    private List<Float> getWeights(final Map<String, Object> params) {
-        if (Objects.isNull(params) || params.isEmpty()) {
-            return List.of();
-        }
-        // get weights, we don't need to check for instance as it's done during validation
-        return ((List<Double>) params.getOrDefault(PARAM_NAME_WEIGHTS, List.of())).stream()
-            .map(Double::floatValue)
-            .collect(Collectors.toUnmodifiableList());
-    }
-
     /**
      * Arithmetic mean method for combining scores.
      * score = (weight1*score1 + weight2*score2 +...+ weightN*scoreN)/(weight1 + weight2 + ... + weightN)
@@ -48,57 +34,24 @@ private List<Float> getWeights(final Map<String, Object> params) {
     @Override
     public float combine(final float[] scores) {
         float combinedScore = 0.0f;
-        float weights = 0;
+        float sumOfWeights = 0;
         for (int indexOfSubQuery = 0; indexOfSubQuery < scores.length; indexOfSubQuery++) {
             float score = scores[indexOfSubQuery];
             if (score >= 0.0) {
-                float weight = getWeightForSubQuery(indexOfSubQuery);
+                float weight = getWeightForSubQuery(weights, indexOfSubQuery);
                 score = score * weight;
                 combinedScore += score;
-                weights += weight;
+                sumOfWeights += weight;
             }
         }
-        if (weights == 0.0f) {
+        if (sumOfWeights == 0.0f) {
             return ZERO_SCORE;
         }
-        return combinedScore / weights;
+        return combinedScore / sumOfWeights;
     }
 
-    private void validateParams(final Map<String, Object> params) {
-        if (Objects.isNull(params) || params.isEmpty()) {
-            return;
-        }
-        // check if only supported params are passed
-        Optional<String> optionalNotSupportedParam = params.keySet()
-            .stream()
-            .filter(paramName -> !SUPPORTED_PARAMS.contains(paramName))
-            .findFirst();
-        if (optionalNotSupportedParam.isPresent()) {
-            throw new IllegalArgumentException(
-                String.format(
-                    Locale.ROOT,
-                    "provided parameter for combination technique is not supported. supported parameters are [%s]",
-                    SUPPORTED_PARAMS.stream().collect(Collectors.joining(","))
-                )
-            );
-        }
-
-        // check param types
-        if (params.keySet().stream().anyMatch(PARAM_NAME_WEIGHTS::equalsIgnoreCase)) {
-            if (!(params.get(PARAM_NAME_WEIGHTS) instanceof List)) {
-                throw new IllegalArgumentException(
-                    String.format(Locale.ROOT, "parameter [%s] must be a collection of numbers", PARAM_NAME_WEIGHTS)
-                );
-            }
-        }
-    }
-
-    /**
-     * Get weight for sub-query based on its index in the hybrid search query. Use user provided weight or 1.0 otherwise
-     * @param indexOfSubQuery 0-based index of sub-query in the Hybrid Search query
-     * @return weight for sub-query, use one that is set in processor/pipeline definition or 1.0 as default
-     */
-    private float getWeightForSubQuery(int indexOfSubQuery) {
-        return indexOfSubQuery < weights.size() ? weights.get(indexOfSubQuery) : 1.0f;
+    @Override
+    Set<String> getSupportedParams() {
+        return SUPPORTED_PARAMS;
     }
 }

src/main/java/org/opensearch/neuralsearch/processor/combination/HarmonicMeanScoreCombinationTechnique.java

@@ -6,19 +6,15 @@
 package org.opensearch.neuralsearch.processor.combination;
 
 import java.util.List;
-import java.util.Locale;
 import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
 import java.util.Set;
-import java.util.stream.Collectors;
 
 /**
- * Abstracts combination of scores based on arithmetic mean method
+ * Abstracts combination of scores based on harmonic mean method
  */
-public class HarmonicMeanScoreCombinationTechnique implements ScoreCombinationTechnique {
+public class HarmonicMeanScoreCombinationTechnique extends AbstractScoreCombinationTechnique implements ScoreCombinationTechnique {
 
-    public static final String TECHNIQUE_NAME = "arithmetic_mean";
+    public static final String TECHNIQUE_NAME = "harmonic_mean";
     public static final String PARAM_NAME_WEIGHTS = "weights";
     private static final Set<String> SUPPORTED_PARAMS = Set.of(PARAM_NAME_WEIGHTS);
     private static final Float ZERO_SCORE = 0.0f;
@@ -29,76 +25,30 @@ public HarmonicMeanScoreCombinationTechnique(final Map<String, Object> params) {
         weights = getWeights(params);
     }
 
-    private List<Float> getWeights(final Map<String, Object> params) {
-        if (Objects.isNull(params) || params.isEmpty()) {
-            return List.of();
-        }
-        // get weights, we don't need to check for instance as it's done during validation
-        return ((List<Double>) params.getOrDefault(PARAM_NAME_WEIGHTS, List.of())).stream()
-            .map(Double::floatValue)
-            .collect(Collectors.toUnmodifiableList());
-    }
-
     /**
-     * Arithmetic mean method for combining scores.
-     * score = (weight1*score1 + weight2*score2 +...+ weightN*scoreN)/(weight1 + weight2 + ... + weightN)
+     * Weighted harmonic mean method for combining scores.
+     * score = sum(weight_1 + .... + weight_n)/sum(weight_1/score_1 + ... + weight_n/score_n)
      *
      * Zero (0.0) scores are excluded from number of scores N
      */
     @Override
     public float combine(final float[] scores) {
-        float combinedScore = 0.0f;
-        float weights = 0;
+        float sumOfWeights = 0;
+        float sumOfHarmonics = 0;
         for (int indexOfSubQuery = 0; indexOfSubQuery < scores.length; indexOfSubQuery++) {
             float score = scores[indexOfSubQuery];
-            if (score >= 0.0) {
-                float weight = getWeightForSubQuery(indexOfSubQuery);
-                score = score * weight;
-                combinedScore += score;
-                weights += weight;
-            }
-        }
-        if (weights == 0.0f) {
-            return ZERO_SCORE;
-        }
-        return combinedScore / weights;
-    }
-
-    private void validateParams(final Map<String, Object> params) {
-        if (Objects.isNull(params) || params.isEmpty()) {
-            return;
-        }
-        // check if only supported params are passed
-        Optional<String> optionalNotSupportedParam = params.keySet()
-            .stream()
-            .filter(paramName -> !SUPPORTED_PARAMS.contains(paramName))
-            .findFirst();
-        if (optionalNotSupportedParam.isPresent()) {
-            throw new IllegalArgumentException(
-                String.format(
-                    Locale.ROOT,
-                    "provided parameter for combination technique is not supported. supported parameters are [%s]",
-                    SUPPORTED_PARAMS.stream().collect(Collectors.joining(","))
-                )
-            );
-        }
-
-        // check param types
-        if (params.keySet().stream().anyMatch(PARAM_NAME_WEIGHTS::equalsIgnoreCase)) {
-            if (!(params.get(PARAM_NAME_WEIGHTS) instanceof List)) {
-                throw new IllegalArgumentException(
-                    String.format(Locale.ROOT, "parameter [%s] must be a collection of numbers", PARAM_NAME_WEIGHTS)
-                );
+            if (score <= 0) {
+                continue;
             }
+            float weightOfSubQuery = getWeightForSubQuery(weights, indexOfSubQuery);
+            sumOfWeights += weightOfSubQuery;
+            sumOfHarmonics += weightOfSubQuery / score;
         }
+        return sumOfHarmonics > 0 ? sumOfWeights / sumOfHarmonics : ZERO_SCORE;
     }
 
-    /**
-     * Get weight for sub-query based on its index in the hybrid search query. Use user provided weight or 1.0 otherwise
-     * @param indexOfSubQuery 0-based index of sub-query in the Hybrid Search query
-     * @return weight for sub-query, use one that is set in processor/pipeline definition or 1.0 as default
-     */
-    private float getWeightForSubQuery(int indexOfSubQuery) {
-        return indexOfSubQuery < weights.size() ? weights.get(indexOfSubQuery) : 1.0f;
+    @Override
+    Set<String> getSupportedParams() {
+        return SUPPORTED_PARAMS;
     }
 }

src/main/java/org/opensearch/neuralsearch/processor/combination/ScoreCombinationFactory.java

@@ -18,7 +18,9 @@ public class ScoreCombinationFactory {
 
     private final Map<String, Function<Map<String, Object>, ScoreCombinationTechnique>> scoreCombinationMethodsMap = Map.of(
         ArithmeticMeanScoreCombinationTechnique.TECHNIQUE_NAME,
-        ArithmeticMeanScoreCombinationTechnique::new
+        ArithmeticMeanScoreCombinationTechnique::new,
+        HarmonicMeanScoreCombinationTechnique.TECHNIQUE_NAME,
+        HarmonicMeanScoreCombinationTechnique::new
     );
 
     /**

src/test/java/org/opensearch/neuralsearch/common/BaseNeuralSearchIT.java

@@ -58,8 +58,8 @@ public abstract class BaseNeuralSearchIT extends OpenSearchSecureRestTestCase {
 
     private static final int DEFAULT_TASK_RESULT_QUERY_INTERVAL_IN_MILLISECOND = 1000;
     private static final String DEFAULT_USER_AGENT = "Kibana";
-    protected static final String NORMALIZATION_METHOD = "min_max";
-    protected static final String COMBINATION_METHOD = "arithmetic_mean";
+    protected static final String DEFAULT_NORMALIZATION_METHOD = "min_max";
+    protected static final String DEFAULT_COMBINATION_METHOD = "arithmetic_mean";
     protected static final String PARAM_NAME_WEIGHTS = "weights";
 
     protected final ClassLoader classLoader = this.getClass().getClassLoader();
@@ -556,7 +556,7 @@ public boolean isUpdateClusterSettings() {
 
     @SneakyThrows
     protected void createSearchPipelineWithResultsPostProcessor(final String pipelineId) {
-        createSearchPipeline(pipelineId, NORMALIZATION_METHOD, COMBINATION_METHOD, Map.of());
+        createSearchPipeline(pipelineId, DEFAULT_NORMALIZATION_METHOD, DEFAULT_COMBINATION_METHOD, Map.of());
     }
 
     @SneakyThrows

src/test/java/org/opensearch/neuralsearch/processor/NormalizationProcessorTests.java

@@ -28,12 +28,12 @@
 import org.opensearch.action.search.SearchPhaseName;
 import org.opensearch.action.search.SearchProgressListener;
 import org.opensearch.action.search.SearchRequest;
-import org.opensearch.common.breaker.CircuitBreaker;
-import org.opensearch.common.breaker.NoopCircuitBreaker;
 import org.opensearch.common.lucene.search.TopDocsAndMaxScore;
 import org.opensearch.common.util.BigArrays;
 import org.opensearch.common.util.concurrent.OpenSearchExecutors;
 import org.opensearch.common.util.concurrent.OpenSearchThreadPoolExecutor;
+import org.opensearch.core.common.breaker.CircuitBreaker;
+import org.opensearch.core.common.breaker.NoopCircuitBreaker;
 import org.opensearch.core.index.shard.ShardId;
 import org.opensearch.neuralsearch.TestUtils;
 import org.opensearch.neuralsearch.processor.combination.ArithmeticMeanScoreCombinationTechnique;