More alignment and cleanup

Signed-off-by: Atri Sharma <atri.jiit@gmail.com>

Author: atris

server/src/main/java/org/opensearch/action/search/StreamQueryPhaseResultConsumer.java

@@ -27,7 +27,6 @@
  * Batch reduction frequency is controlled by per-mode multipliers:
  * - NO_SCORING: Immediate reduction (batch size = 1) for fastest time-to-first-byte
  * - SCORED_UNSORTED: Small batches (minBatchReduceSize * 2)
- * - CONFIDENCE_BASED: Moderate batches (minBatchReduceSize * 3)
  * - SCORED_SORTED: Larger batches (minBatchReduceSize * 10)
  *
  * These multipliers are applied to the base batch reduce size (typically 5) to determine
@@ -87,20 +86,19 @@ public StreamQueryPhaseResultConsumer(
     @Override
     int getBatchReduceSize(int requestBatchedReduceSize, int minBatchReduceSize) {
         // Handle null during construction (parent constructor calls this before our constructor body runs)
-        if (scoringMode == null) {
-            return super.getBatchReduceSize(requestBatchedReduceSize, minBatchReduceSize * 10);
+        // Use SCORED_UNSORTED as default to match other defaults
+        StreamingSearchMode mode = scoringMode;
+        if (mode == null) {
+            mode = StreamingSearchMode.SCORED_UNSORTED;
         }
 
-        switch (scoringMode) {
+        switch (mode) {
             case NO_SCORING:
                 // Reduce immediately for fastest TTFB
                 return Math.min(requestBatchedReduceSize, 1);
             case SCORED_UNSORTED:
                 // Small batches for quick emission without sorting overhead
                 return super.getBatchReduceSize(requestBatchedReduceSize, minBatchReduceSize * 2);
-            case CONFIDENCE_BASED:
-                // Moderate batching for progressive emission with confidence
-                return super.getBatchReduceSize(requestBatchedReduceSize, minBatchReduceSize * 3);
             case SCORED_SORTED:
                 // Higher batch size to collect more results before reducing (sorting is expensive)
                 return super.getBatchReduceSize(requestBatchedReduceSize, minBatchReduceSize * 10);

server/src/main/java/org/opensearch/action/search/StreamingSearchProgressListener.java

@@ -22,7 +22,7 @@
 
 /**
  * SearchProgressListener implementation for streaming search with scoring.
- * Computes partial search results when confidence thresholds are met.
+ * Computes partial search results at configured intervals and milestones.
  *
  * @opensearch.internal
  */
@@ -96,7 +96,7 @@ protected void onPartialReduceWithTopDocs(
             collectPartialResponse(partialResponse);
 
             int count = streamEmissions.incrementAndGet();
-            logger.info("Computed streaming partial #{} with {} docs from {} shards", count, topDocs.scoreDocs.length, shards.size());
+            logger.debug("Computed streaming partial #{} with {} docs from {} shards", count, topDocs.scoreDocs.length, shards.size());
 
         } catch (Exception e) {
             logger.error("Failed to send partial TopDocs", e);
@@ -113,7 +113,7 @@ private void collectPartialResponse(SearchResponse partialResponse) {
 
     @Override
     protected void onFinalReduce(List<SearchShard> shards, TotalHits totalHits, InternalAggregations aggs, int reducePhase) {
-        logger.info(
+        logger.debug(
             "Final reduce: {} total hits from {} shards, {} partial computations",
             totalHits.value(),
             shards.size(),

server/src/main/java/org/opensearch/action/search/StreamingSearchProgressMessage.java

@@ -59,7 +59,6 @@ public static Milestone fromProgress(float progress) {
     private final long requestId;
     private final Milestone milestone;
     private final TopDocs topDocs;
-    private final float confidence;
     private final ProgressStatistics statistics;
     private final boolean isFinal;
 
@@ -70,7 +69,6 @@ public StreamingSearchProgressMessage() {
         this.requestId = -1;
         this.milestone = Milestone.INITIAL;
         this.topDocs = null;
-        this.confidence = 0.0f;
         this.statistics = null;
         this.isFinal = false;
     }
@@ -81,7 +79,6 @@ public StreamingSearchProgressMessage(
         long requestId,
         Milestone milestone,
         TopDocs topDocs,
-        float confidence,
         ProgressStatistics statistics,
         boolean isFinal
     ) {
@@ -90,7 +87,6 @@ public StreamingSearchProgressMessage(
         this.requestId = requestId;
         this.milestone = milestone;
         this.topDocs = topDocs;
-        this.confidence = confidence;
         this.statistics = statistics;
         this.isFinal = isFinal;
     }
@@ -113,7 +109,6 @@ public StreamingSearchProgressMessage(StreamInput in) throws IOException {
         }
         this.topDocs = new TopDocs(new TotalHits(totalHits, TotalHits.Relation.EQUAL_TO), scoreDocs);
 
-        this.confidence = in.readFloat();
         this.statistics = new ProgressStatistics(in);
         this.isFinal = in.readBoolean();
     }
@@ -133,7 +128,6 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeFloat(doc.score);
         }
 
-        out.writeFloat(confidence);
         statistics.writeTo(out);
         out.writeBoolean(isFinal);
     }
@@ -158,10 +152,6 @@ public TopDocs getTopDocs() {
         return topDocs;
     }
 
-    public float getConfidence() {
-        return confidence;
-    }
-
     public ProgressStatistics getStatistics() {
         return statistics;
     }
@@ -270,7 +260,6 @@ public static class Builder {
         private long requestId;
         private Milestone milestone;
         private TopDocs topDocs;
-        private float confidence = 0.0f;
         private ProgressStatistics statistics;
         private boolean isFinal = false;
 
@@ -299,11 +288,6 @@ public Builder topDocs(TopDocs topDocs) {
             return this;
         }
 
-        public Builder confidence(float confidence) {
-            this.confidence = confidence;
-            return this;
-        }
-
         public Builder statistics(ProgressStatistics statistics) {
             this.statistics = statistics;
             return this;
@@ -315,16 +299,7 @@ public Builder isFinal(boolean isFinal) {
         }
 
         public StreamingSearchProgressMessage build() {
-            return new StreamingSearchProgressMessage(
-                shardTarget,
-                shardIndex,
-                requestId,
-                milestone,
-                topDocs,
-                confidence,
-                statistics,
-                isFinal
-            );
+            return new StreamingSearchProgressMessage(shardTarget, shardIndex, requestId, milestone, topDocs, statistics, isFinal);
         }
     }
 }

server/src/main/java/org/opensearch/action/search/StreamingSearchResponseListener.java

@@ -58,7 +58,7 @@ public void onPartialResponse(SearchResponse partialResponse) {
         // Track TTFB - first partial result delivery time
         if (count == 1 && partialResponse.getHits() != null) {
             int numHits = partialResponse.getHits().getHits().length;
-            logger.info("First partial result delivered with {} hits", numHits);
+            logger.debug("First partial result delivered with {} hits", numHits);
         }
     }
 
@@ -90,7 +90,7 @@ private void logPartialResponse(SearchResponse partialResponse, int count) {
             int numHits = partialResponse.getHits().getHits().length;
             long totalHits = partialResponse.getHits().getTotalHits().value();
 
-            logger.info("Streaming partial result #{}: {} hits, total: {}", count, numHits, totalHits);
+            logger.debug("Streaming partial result #{}: {} hits, total: {}", count, numHits, totalHits);
 
             if (logger.isDebugEnabled() && numHits > 0) {
                 float topScore = partialResponse.getHits().getHits()[0].getScore();
@@ -102,13 +102,13 @@ private void logPartialResponse(SearchResponse partialResponse, int count) {
     private void logStreamingSummary(SearchResponse finalResponse) {
         int totalPartials = partialCount.get();
         if (totalPartials > 0) {
-            logger.info("Streaming search complete: {} partial computations", totalPartials);
+            logger.debug("Streaming search complete: {} partial computations", totalPartials);
 
             if (!partialResponses.isEmpty()) {
                 long totalDocsProcessed = partialResponses.stream()
                     .mapToLong(r -> r.getHits() != null ? r.getHits().getHits().length : 0)
                     .sum();
-                logger.info("Processed {} docs across {} partial emissions", totalDocsProcessed, partialResponses.size());
+                logger.debug("Processed {} docs across {} partial emissions", totalDocsProcessed, partialResponses.size());
             }
         } else {
             logger.debug("No partial computations performed");

server/src/main/java/org/opensearch/action/search/TransportSearchAction.java

@@ -82,6 +82,7 @@
 import org.opensearch.search.pipeline.SearchPipelineService;
 import org.opensearch.search.profile.ProfileShardResult;
 import org.opensearch.search.profile.SearchProfileShardResults;
+import org.opensearch.search.query.StreamingSearchMode;
 import org.opensearch.search.slice.SliceBuilder;
 import org.opensearch.tasks.CancellableTask;
 import org.opensearch.tasks.Task;
@@ -135,7 +136,8 @@
  * @opensearch.internal
  */
 public class TransportSearchAction extends HandledTransportAction<SearchRequest, SearchResponse> {
-    // Streaming search integrated via streamingSearchMode in SearchRequest
+    // Streaming is integrated via SearchAction (TransportSearchAction). When stream transport is available,
+    // register its request handler here. A separate StreamSearchAction registration is not required.
 
     /** The maximum number of shards for a single search request. */
     public static final Setting<Long> SHARD_COUNT_LIMIT_SETTING = Setting.longSetting(
@@ -1269,7 +1271,13 @@ AbstractSearchAsyncAction<? extends SearchPhaseResult> searchAsyncAction(
                 tracer
             );
         } else {
-            final boolean isStreamingRequest = searchRequest.getStreamingSearchMode() != null
+            // Set default streaming mode if flag is present but mode is null
+            if (searchRequest.isStreamingScoring() && searchRequest.getStreamingSearchMode() == null) {
+                searchRequest.setStreamingSearchMode(StreamingSearchMode.SCORED_UNSORTED.toString());
+            }
+
+            // Consider streaming if either flag is set or mode is specified
+            final boolean isStreamingRequest = (searchRequest.isStreamingScoring() || searchRequest.getStreamingSearchMode() != null)
                 && (searchRequest.source() == null || searchRequest.source().size() > 0);
 
             final SearchProgressListener progressListener = isStreamingRequest

server/src/main/java/org/opensearch/search/SearchService.java

@@ -821,7 +821,7 @@ private SearchPhaseResult executeQueryPhase(
                     // FALLBACK: If this is a streaming search but no mode is set, use a default mode
                     // This handles the case where ShardSearchRequest is created without copying streaming fields
                     if (isStreamSearch) {
-                        context.setStreamingMode(StreamingSearchMode.NO_SCORING);
+                        context.setStreamingMode(StreamingSearchMode.SCORED_UNSORTED);
                     }
                 }
             }

server/src/main/java/org/opensearch/search/builder/StreamingSearchParameters.java

@@ -26,15 +26,13 @@ public class StreamingSearchParameters implements Writeable, ToXContent {
 
     public static final String STREAMING_FIELD = "streaming";
     public static final String ENABLED_FIELD = "enabled";
-    public static final String CONFIDENCE_FIELD = "confidence";
     public static final String BATCH_SIZE_FIELD = "batch_size";
     public static final String EMISSION_INTERVAL_FIELD = "emission_interval";
     public static final String MIN_DOCS_FIELD = "min_docs";
     public static final String ADAPTIVE_BATCHING_FIELD = "adaptive_batching";
     public static final String MILESTONES_FIELD = "milestones";
 
     private boolean enabled = false;
-    private float initialConfidence = 0.99f;
     private int batchSize = 10;
     private int emissionIntervalMillis = 100;
     private int minDocsForStreaming = 5;
@@ -45,7 +43,6 @@ public StreamingSearchParameters() {}
 
     public StreamingSearchParameters(StreamInput in) throws IOException {
         this.enabled = in.readBoolean();
-        this.initialConfidence = in.readFloat();
         this.batchSize = in.readVInt();
         this.emissionIntervalMillis = in.readVInt();
         this.minDocsForStreaming = in.readVInt();
@@ -56,7 +53,6 @@ public StreamingSearchParameters(StreamInput in) throws IOException {
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         out.writeBoolean(enabled);
-        out.writeFloat(initialConfidence);
         out.writeVInt(batchSize);
         out.writeVInt(emissionIntervalMillis);
         out.writeVInt(minDocsForStreaming);
@@ -68,7 +64,6 @@ public void writeTo(StreamOutput out) throws IOException {
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject(STREAMING_FIELD);
         builder.field(ENABLED_FIELD, enabled);
-        builder.field(CONFIDENCE_FIELD, initialConfidence);
         builder.field(BATCH_SIZE_FIELD, batchSize);
         builder.field(EMISSION_INTERVAL_FIELD, emissionIntervalMillis);
         builder.field(MIN_DOCS_FIELD, minDocsForStreaming);
@@ -90,8 +85,6 @@ public static StreamingSearchParameters fromXContent(XContentParser parser) thro
             } else if (token.isValue()) {
                 if (ENABLED_FIELD.equals(currentFieldName)) {
                     params.enabled = parser.booleanValue();
-                } else if (CONFIDENCE_FIELD.equals(currentFieldName)) {
-                    params.initialConfidence = parser.floatValue();
                 } else if (BATCH_SIZE_FIELD.equals(currentFieldName)) {
                     params.batchSize = parser.intValue();
                 } else if (EMISSION_INTERVAL_FIELD.equals(currentFieldName)) {
@@ -120,18 +113,6 @@ public StreamingSearchParameters enabled(boolean enabled) {
         return this;
     }
 
-    public float getInitialConfidence() {
-        return initialConfidence;
-    }
-
-    public StreamingSearchParameters initialConfidence(float confidence) {
-        if (confidence <= 0.0f || confidence > 1.0f) {
-            throw new IllegalArgumentException("Confidence must be between 0 and 1");
-        }
-        this.initialConfidence = confidence;
-        return this;
-    }
-
     public int getBatchSize() {
         return batchSize;
     }
@@ -192,7 +173,6 @@ public boolean equals(Object o) {
         if (o == null || getClass() != o.getClass()) return false;
         StreamingSearchParameters that = (StreamingSearchParameters) o;
         return enabled == that.enabled
-            && Float.compare(that.initialConfidence, initialConfidence) == 0
             && batchSize == that.batchSize
             && emissionIntervalMillis == that.emissionIntervalMillis
             && minDocsForStreaming == that.minDocsForStreaming
@@ -202,24 +182,14 @@ public boolean equals(Object o) {
 
     @Override
     public int hashCode() {
-        return Objects.hash(
-            enabled,
-            initialConfidence,
-            batchSize,
-            emissionIntervalMillis,
-            minDocsForStreaming,
-            adaptiveBatching,
-            useMilestones
-        );
+        return Objects.hash(enabled, batchSize, emissionIntervalMillis, minDocsForStreaming, adaptiveBatching, useMilestones);
     }
 
     @Override
     public String toString() {
         return "StreamingSearchParameters{"
             + "enabled="
             + enabled
-            + ", confidence="
-            + initialConfidence
             + ", batchSize="
             + batchSize
             + ", emissionInterval="

server/src/main/java/org/opensearch/search/internal/SearchContext.java

@@ -580,4 +580,16 @@ public boolean isStreamingSearch() {
     public int getStreamingBatchSize() {
         return 10;
     }
+
+    public org.opensearch.common.unit.TimeValue getStreamingTimeInterval() {
+        return org.opensearch.common.unit.TimeValue.timeValueMillis(100);
+    }
+
+    public Boolean getStreamingFirstHitImmediate() {
+        return true;
+    }
+
+    public Boolean getStreamingEnableCoalescing() {
+        return true;
+    }
 }