[RRF] Read rank_constant from the combination clause, not combination.parameters

BEHAVIOR FIX on top of opensearch-project#1933. FusionSpec.fromScoreRankerProcessor read
`combination.parameters.rank_constant`, but RRFProcessorFactory — the ground
truth for the score-ranker-processor shape — reads `rank_constant` directly off
the `combination` clause and rejects anything but `weights` under `parameters`.
Two consequences, both caught by an integration test:

  - Every user-supplied rank_constant was silently ignored, so fused mode always
    ranked with the default 60.
  - The shape FusionSpec documented and the IT fixture emitted is rejected by the
    real processor with HTTP 400 "provided parameter for combination technique is
    not supported. supported parameters are [weights]", so a fused query could
    not even reuse an existing RRF pipeline.

Changes:
  - Read rank_constant off the combination clause via the shared
    RRFScoreNormalizer.resolveRankConstant, so fused mode accepts, rejects and
    reports exactly what classic does (absent -> 60, non-integer -> "must be an
    integer", outside [1, 10000] -> the range error). Reading it directly here
    would silently accept a negative, oversized or fractional value.
  - Reject rank_constant under `parameters` with a clear message rather than
    falling back to 60 and mis-ranking every query for a user who put it in the
    wrong place.
  - Report the normalization technique for the rrf shape instead of pinning it to
    "none". The score-ranker-processor has no normalization clause so this still
    resolves to "none", but an inline fusion block can carry one, and reporting it
    lets the caller reject the contradictory pairing instead of silently dropping
    what the user asked for.
  - Correct the two javadocs that documented the wrong location, and emit
    rank_constant in the right place in BaseNeuralSearchIT.createRRFSearchPipeline
    (new optional overload; the 3-arg form delegates with null).
Signed-off-by: Daniel Widdis <widdis@gmail.com>

Author: dbwiddis

src/main/java/org/opensearch/neuralsearch/query/FusionSpec.java

@@ -17,6 +17,7 @@
 import org.opensearch.neuralsearch.processor.combination.ArithmeticMeanScoreCombinationTechnique;
 import org.opensearch.neuralsearch.processor.combination.RRFScoreCombinationTechnique;
 import org.opensearch.neuralsearch.processor.normalization.MinMaxScoreNormalizationTechnique;
+import org.opensearch.neuralsearch.processor.normalization.RRFScoreNormalizer;
 
 /**
  * Immutable, resolved fusion configuration for the {@code hybrid} query resolver (fused) mode — the normalization +
@@ -33,8 +34,9 @@
  * <ul>
  *   <li>{@code normalization-processor}: {@code normalization.technique} (min_max|l2|z_score) +
  *       {@code combination.technique} (arithmetic_mean) + optional {@code combination.parameters.weights}.</li>
- *   <li>{@code score-ranker-processor}: {@code combination.technique = rrf} +
- *       {@code combination.parameters.rank_constant}. RRF is rank-based (no normalization clause).</li>
+ *   <li>{@code score-ranker-processor}: {@code combination.technique = rrf} + {@code combination.rank_constant} (on the
+ *       combination clause itself, NOT under {@code parameters} — that is where {@code RRFProcessorFactory} reads it) +
+ *       optional {@code combination.parameters.weights}. RRF is rank-based (no normalization clause).</li>
  * </ul>
  */
 @Getter(AccessLevel.PACKAGE)
@@ -48,7 +50,8 @@ public final class FusionSpec {
     static final String NORMALIZATION_NONE = "none";
     static final String NORMALIZATION_MIN_MAX = MinMaxScoreNormalizationTechnique.TECHNIQUE_NAME;
 
-    static final int DEFAULT_RANK_CONSTANT = 60;
+    // Sourced from the shared normalizer rather than redeclared, so fused mode and classic cannot drift apart.
+    static final int DEFAULT_RANK_CONSTANT = RRFScoreNormalizer.DEFAULT_RANK_CONSTANT;
 
     // Config-map keys (shared by the normalization-processor and score-ranker-processor definitions)
     private static final String PHASE_RESULTS_PROCESSORS_KEY = "phase_results_processors";
@@ -57,7 +60,7 @@ public final class FusionSpec {
     private static final String TECHNIQUE_KEY = "technique";
     private static final String PARAMETERS_KEY = "parameters";
     private static final String WEIGHTS_KEY = "weights";
-    private static final String RANK_CONSTANT_KEY = "rank_constant";
+    private static final String RANK_CONSTANT_KEY = RRFScoreNormalizer.PARAM_NAME_RANK_CONSTANT;
 
     private final String combinationTechnique; // rrf | arithmetic_mean
     private final String normalizationTechnique; // none | min_max | z_score | l2
@@ -108,7 +111,7 @@ static FusionSpec fromPipelineConfig(Map<String, Object> pipelineConfig) {
     /**
      * Read a {@link FusionSpec} from an inline {@code fusion} block on the query body (precedence step 1: inline wins
      * over the attached pipeline). The block mirrors the processor JSON verbatim —
-     * {@code {normalization: {technique}, combination: {technique, parameters: {weights | rank_constant}}}} — so this
+     * {@code {normalization: {technique}, combination: {technique, rank_constant, parameters: {weights}}}} — so this
      * reuses the pipeline-config parsing. {@code combination.technique: rrf} routes to the rank-constant shape.
      *
      * @param fusionConfig the parsed inline fusion map (nullable)
@@ -130,13 +133,7 @@ static FusionSpec fromInlineFusion(Map<String, Object> fusionConfig) {
 
     @SuppressWarnings("unchecked")
     private static FusionSpec fromNormalizationProcessor(Map<String, Object> config) {
-        String normalization = NORMALIZATION_MIN_MAX;
-        if (config.get(NORMALIZATION_CLAUSE) instanceof Map) {
-            Object technique = ((Map<String, Object>) config.get(NORMALIZATION_CLAUSE)).get(TECHNIQUE_KEY);
-            if (Objects.nonNull(technique)) {
-                normalization = technique.toString().toLowerCase(Locale.ROOT);
-            }
-        }
+        String normalization = readNormalizationTechnique(config, NORMALIZATION_MIN_MAX);
         String combination = TECHNIQUE_ARITHMETIC_MEAN;
         float[] weights = new float[0];
         if (config.get(COMBINATION_CLAUSE) instanceof Map) {
@@ -156,15 +153,52 @@ private static FusionSpec fromScoreRankerProcessor(Map<String, Object> config) {
         float[] weights = new float[0];
         if (config.get(COMBINATION_CLAUSE) instanceof Map) {
             Map<String, Object> combinationClause = (Map<String, Object>) config.get(COMBINATION_CLAUSE);
-            if (combinationClause.get(PARAMETERS_KEY) instanceof Map) {
-                Map<String, Object> parameters = (Map<String, Object>) combinationClause.get(PARAMETERS_KEY);
-                if (parameters.get(RANK_CONSTANT_KEY) instanceof Number) {
-                    rankConstant = ((Number) parameters.get(RANK_CONSTANT_KEY)).intValue();
-                }
-            }
+            rejectRankConstantUnderParameters(combinationClause);
+            // rank_constant sits on the combination clause itself, which is where RRFProcessorFactory reads it — NOT
+            // under `parameters`, whose only supported key is `weights`. Delegating to the shared resolver makes fused
+            // mode accept, reject and report exactly what classic does: absent -> default, non-integer -> "must be an
+            // integer", outside [1, 10000] -> the range error. Reading the value directly here would silently accept a
+            // negative, oversized or fractional rank constant that the score-ranker-processor rejects.
+            rankConstant = RRFScoreNormalizer.resolveRankConstant(combinationClause);
             weights = readWeights(combinationClause);
         }
-        return new FusionSpec(TECHNIQUE_RRF, NORMALIZATION_NONE, rankConstant, weights);
+        // RRF is rank based, so the score-ranker-processor has no normalization clause and this resolves to "none".
+        // An inline fusion block can still carry one, and it is reported rather than dropped so the caller's technique
+        // check rejects the contradictory pairing instead of silently ignoring what the user asked for.
+        return new FusionSpec(TECHNIQUE_RRF, readNormalizationTechnique(config, NORMALIZATION_NONE), rankConstant, weights);
+    }
+
+    /**
+     * {@code rank_constant} under {@code combination.parameters} is a config error, not a place we also look: the
+     * score-ranker-processor rejects it there ("supported parameters are [weights]"). Fused mode rejects it too, rather
+     * than silently falling back to the default 60 and mis-ranking every query for a user who put it in the wrong place.
+     */
+    @SuppressWarnings("unchecked")
+    private static void rejectRankConstantUnderParameters(Map<String, Object> combinationClause) {
+        if ((combinationClause.get(PARAMETERS_KEY) instanceof Map) == false) {
+            return;
+        }
+        if (((Map<String, Object>) combinationClause.get(PARAMETERS_KEY)).containsKey(RANK_CONSTANT_KEY)) {
+            throw new IllegalArgumentException(
+                String.format(
+                    Locale.ROOT,
+                    "[%s] must be set on the [%s] clause, not under [%s]; supported parameters are [%s]",
+                    RANK_CONSTANT_KEY,
+                    COMBINATION_CLAUSE,
+                    PARAMETERS_KEY,
+                    WEIGHTS_KEY
+                )
+            );
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    private static String readNormalizationTechnique(Map<String, Object> config, String defaultTechnique) {
+        if ((config.get(NORMALIZATION_CLAUSE) instanceof Map) == false) {
+            return defaultTechnique;
+        }
+        Object technique = ((Map<String, Object>) config.get(NORMALIZATION_CLAUSE)).get(TECHNIQUE_KEY);
+        return Objects.isNull(technique) ? defaultTechnique : technique.toString().toLowerCase(Locale.ROOT);
     }
 
     @SuppressWarnings("unchecked")

src/test/java/org/opensearch/neuralsearch/query/FusionSpecTests.java

@@ -26,14 +26,58 @@ public void testFromInlineFusion_whenNormalizationCombination_thenParsed() {
     }
 
     public void testFromInlineFusion_whenRrf_thenRankConstantAndNoNormalization() {
-        Map<String, Object> inline = Map.of("combination", Map.of("technique", "rrf", "parameters", Map.of("rank_constant", 42)));
+        Map<String, Object> inline = Map.of("combination", Map.of("technique", "rrf", "rank_constant", 42));
         FusionSpec spec = FusionSpec.fromInlineFusion(inline);
         assertNotNull(spec);
         assertEquals(FusionSpec.TECHNIQUE_RRF, spec.combinationTechnique());
         assertEquals(FusionSpec.NORMALIZATION_NONE, spec.normalizationTechnique());
         assertEquals(42, spec.rankConstant());
     }
 
+    public void testFromInlineFusion_whenRrfWithNormalizationClause_thenNormalizationReported() {
+        // RRF takes no normalization technique, but an inline block can still carry one. It is reported rather than
+        // dropped so the caller's technique check can reject the contradictory pairing.
+        Map<String, Object> inline = Map.of("normalization", Map.of("technique", "min_max"), "combination", Map.of("technique", "rrf"));
+        FusionSpec spec = FusionSpec.fromInlineFusion(inline);
+        assertNotNull(spec);
+        assertEquals(FusionSpec.TECHNIQUE_RRF, spec.combinationTechnique());
+        assertEquals(FusionSpec.NORMALIZATION_MIN_MAX, spec.normalizationTechnique());
+    }
+
+    public void testFromInlineFusion_whenRrfRankConstantInvalid_thenRejected() {
+        // Resolved through the shared validator, so fused mode rejects exactly what the score-ranker-processor rejects.
+        assertThrows(
+            IllegalArgumentException.class,
+            () -> FusionSpec.fromInlineFusion(Map.of("combination", Map.of("technique", "rrf", "rank_constant", 0)))
+        );
+        assertThrows(
+            IllegalArgumentException.class,
+            () -> FusionSpec.fromInlineFusion(Map.of("combination", Map.of("technique", "rrf", "rank_constant", 10001)))
+        );
+        assertThrows(
+            IllegalArgumentException.class,
+            () -> FusionSpec.fromInlineFusion(Map.of("combination", Map.of("technique", "rrf", "rank_constant", "not-a-number")))
+        );
+    }
+
+    public void testFromInlineFusion_whenRankConstantUnderParameters_thenRejected() {
+        // The score-ranker-processor reads rank_constant off the combination clause and rejects it under `parameters`
+        // ("supported parameters are [weights]"). Fused mode must reject it the same way rather than silently defaulting
+        // to 60 and mis-ranking every query.
+        IllegalArgumentException e = assertThrows(
+            IllegalArgumentException.class,
+            () -> FusionSpec.fromInlineFusion(Map.of("combination", Map.of("technique", "rrf", "parameters", Map.of("rank_constant", 42))))
+        );
+        assertTrue(e.getMessage().contains("must be set on the [combination] clause"));
+    }
+
+    public void testFromInlineFusion_whenRrfWithoutRankConstant_thenDefault() {
+        FusionSpec spec = FusionSpec.fromInlineFusion(Map.of("combination", Map.of("technique", "rrf")));
+        assertNotNull(spec);
+        assertEquals(FusionSpec.TECHNIQUE_RRF, spec.combinationTechnique());
+        assertEquals(FusionSpec.DEFAULT_RANK_CONSTANT, spec.rankConstant());
+    }
+
     public void testFromInlineFusion_whenDefaults_thenMinMaxArithmeticMean() {
         // An empty inline block resolves to the min_max + arithmetic_mean defaults.
         FusionSpec spec = FusionSpec.fromInlineFusion(Map.of());
@@ -67,12 +111,7 @@ public void testFromPipelineConfig_whenNormalizationProcessor_thenParsed() {
     public void testFromPipelineConfig_whenScoreRankerProcessor_thenRrf() {
         Map<String, Object> pipelineConfig = Map.of(
             "phase_results_processors",
-            List.of(
-                Map.of(
-                    "score-ranker-processor",
-                    Map.of("combination", Map.of("technique", "rrf", "parameters", Map.of("rank_constant", 10)))
-                )
-            )
+            List.of(Map.of("score-ranker-processor", Map.of("combination", Map.of("technique", "rrf", "rank_constant", 10))))
         );
         FusionSpec spec = FusionSpec.fromPipelineConfig(pipelineConfig);
         assertNotNull(spec);

src/test/java/org/opensearch/neuralsearch/query/HybridQueryBuilderTests.java

@@ -659,7 +659,7 @@ public void testDoRewriteFused_whenUnsupportedTechnique_thenFailsFast() {
     public void testDoRewriteFused_whenRrf_thenFailsFast() {
         setUpClusterService();
         HybridQueryBuilder builder = fusedBuilder(
-            new HashMap<>(Map.of("combination", Map.of("technique", "rrf", "parameters", Map.of("rank_constant", 60))))
+            new HashMap<>(Map.of("combination", Map.of("technique", "rrf", "rank_constant", 60)))
         );
         QueryCoordinatorContext ctx = coordinatorContextFor(builder);
         IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> builder.doRewrite(ctx));

src/testFixtures/java/org/opensearch/neuralsearch/BaseNeuralSearchIT.java

@@ -2766,15 +2766,34 @@ protected void createDefaultRRFSearchPipeline() {
 
     @SneakyThrows
     protected void createRRFSearchPipeline(final String pipelineName, final List<Double> weights, boolean addExplainResponseProcessor) {
+        createRRFSearchPipeline(pipelineName, weights, null, addExplainResponseProcessor);
+    }
+
+    /**
+     * @param rankConstant written as {@code combination.rank_constant} when non-null — the location
+     *                     {@code RRFProcessorFactory} reads, which is the combination clause itself and NOT
+     *                     {@code parameters} (whose only supported key is {@code weights}). Null omits the key so the
+     *                     processor applies its own default.
+     */
+    @SneakyThrows
+    protected void createRRFSearchPipeline(
+        final String pipelineName,
+        final List<Double> weights,
+        final Integer rankConstant,
+        boolean addExplainResponseProcessor
+    ) {
         XContentBuilder builder = XContentFactory.jsonBuilder()
             .startObject()
             .field("description", "Post processor for hybrid search")
             .startArray("phase_results_processors")
             .startObject()
             .startObject("score-ranker-processor")
             .startObject("combination")
-            .field("technique", "rrf")
-            .startObject("parameters");
+            .field("technique", "rrf");
+        if (rankConstant != null) {
+            builder.field("rank_constant", rankConstant);
+        }
+        builder.startObject("parameters");
         if (weights.size() > 0) {
             builder.startArray("weights");
             for (Double weight : weights) {