Cleanup

Signed-off-by: Atri Sharma <atrisharma@Atris-Mac-Studio.local>

Author: Atri Sharma

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

@@ -35,7 +35,6 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.lucene.search.TopDocs;
-import org.apache.lucene.search.TotalHits;
 import org.opensearch.common.lease.Releasable;
 import org.opensearch.common.lease.Releasables;
 import org.opensearch.common.lucene.search.TopDocsAndMaxScore;

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

@@ -355,7 +355,7 @@ public XContentBuilder innerToXContent(XContentBuilder builder, Params params) t
         if (getNumReducePhases() != 1) {
             builder.field(NUM_REDUCE_PHASES.getPreferredName(), getNumReducePhases());
         }
-        
+
         RestActions.buildBroadcastShardsHeader(
             builder,
             params,

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

@@ -16,14 +16,11 @@
 
 /**
  * Query phase result consumer for streaming search.
- * Supports progressive batch reduction with a fixed batch policy suitable for unsorted streaming.
  *
  * @opensearch.internal
  */
 public class StreamQueryPhaseResultConsumer extends QueryPhaseResultConsumer {
 
-
-
     /**
      * Creates a streaming query phase result consumer.
      */
@@ -49,18 +46,9 @@ public StreamQueryPhaseResultConsumer(
         );
     }
 
-    /**
-     * Controls partial reduction frequency.
-     * With NO_SCORING streaming, we reduce immediately for fastest TTFB.
-     *
-     * @param requestBatchedReduceSize request batch size
-     * @param minBatchReduceSize       minimum batch size
-     */
     @Override
     int getBatchReduceSize(int requestBatchedReduceSize, int minBatchReduceSize) {
         // Reduce immediately for fastest TTFB
         return Math.min(requestBatchedReduceSize, 1);
     }
 }
-
-

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

@@ -73,10 +73,6 @@ public class StreamSearchQueryThenFetchAsyncAction extends SearchQueryThenFetchA
         this.logger = logger;
     }
 
-    /**
-     * Override the extension point to create streaming listeners instead of regular
-     * listeners
-     */
     @Override
     SearchActionListener<SearchPhaseResult> createShardActionListener(
         final SearchShardTarget shard,
@@ -145,14 +141,9 @@ protected void onStreamResult(SearchPhaseResult result, SearchShardIterator shar
         results.consumeResult(result, next);
     }
 
-    /**
-     * Override onShardResult to handle streaming search results safely.
-     * This prevents the "topDocs already consumed" error when processing
-     * multiple streaming results from the same shard.
-     */
     @Override
     protected void onShardResult(SearchPhaseResult result, SearchShardIterator shardIt) {
-        // Safety log: track final shard response receipt in coordinator
+        // Trace final shard responses to diagnose coordinator sequencing.
         if (logger.isTraceEnabled()) {
             logger.trace(
                 "COORDINATOR: received final shard result from shard={}, target={}, totalOps={}, expectedOps={}",
@@ -162,14 +153,9 @@ protected void onShardResult(SearchPhaseResult result, SearchShardIterator shard
                 expectedTotalOps
             );
         }
-        // Always delegate to the parent to ensure shard accounting and phase
-        // transitions.
         super.onShardResult(result, shardIt);
     }
 
-    /**
-     * Override successful shard execution to handle stream result synchronization
-     */
     @Override
     void successfulShardExecution(SearchShardIterator shardsIt) {
         final int remainingOpsOnIterator;
@@ -194,14 +180,8 @@ void successfulShardExecution(SearchShardIterator shardsIt) {
         }
     }
 
-    /**
-     * Handle successful stream execution callback
-     * Since partials are no longer fed into the reducer, this callback is not
-     * needed for coordination.
-     */
     private void successfulStreamExecution() {
-        // No-op: partials are bypassed from reducer, completion is handled by
-        // successfulShardExecution only
+        // No-op.
     }
 
 }

server/src/main/java/org/opensearch/rest/action/search/RestSearchAction.java

@@ -39,7 +39,6 @@
 import org.opensearch.action.search.SearchAction;
 import org.opensearch.action.search.SearchContextId;
 import org.opensearch.action.search.SearchRequest;
-import org.opensearch.action.search.SearchResponse;
 import org.opensearch.action.search.StreamSearchAction;
 import org.opensearch.action.support.IndicesOptions;
 import org.opensearch.common.Booleans;

server/src/main/java/org/opensearch/search/aggregations/bucket/terms/StreamNumericTermsAggregator.java

@@ -32,8 +32,8 @@
 import org.opensearch.search.aggregations.bucket.LocalBucketCountThresholds;
 import org.opensearch.search.aggregations.support.ValuesSource;
 import org.opensearch.search.internal.SearchContext;
-import org.opensearch.search.streaming.StreamingCostMetrics;
 import org.opensearch.search.streaming.StreamingCostEstimable;
+import org.opensearch.search.streaming.StreamingCostMetrics;
 
 import java.io.IOException;
 import java.util.ArrayList;

server/src/main/java/org/opensearch/search/aggregations/bucket/terms/StreamStringTermsAggregator.java

@@ -514,9 +514,7 @@ public StreamingCostMetrics estimateStreamingCost(SearchContext context) {
                 return StreamingCostMetrics.nonStreamable();
             }
 
-            // Calculate metrics (simplified for now to match StreamNumericTermsAggregator
-            // pattern essentially)
-            // But we need total bucket count from ordinals
+            // Estimate streaming cost using segment ordinals.
             ValuesSource.Bytes.WithOrdinals ordinalValuesSource = (ValuesSource.Bytes.WithOrdinals) valuesSource;
             List<LeafReaderContext> leaves = context.searcher().getIndexReader().leaves();
             long maxCardinality = 0;

server/src/main/java/org/opensearch/search/aggregations/metrics/StreamCardinalityAggregator.java

@@ -16,8 +16,8 @@
 import org.opensearch.search.aggregations.support.ValuesSource;
 import org.opensearch.search.aggregations.support.ValuesSourceConfig;
 import org.opensearch.search.internal.SearchContext;
-import org.opensearch.search.streaming.StreamingCostMetrics;
 import org.opensearch.search.streaming.StreamingCostEstimable;
+import org.opensearch.search.streaming.StreamingCostMetrics;
 
 import java.io.IOException;
 import java.util.List;

server/src/main/java/org/opensearch/search/query/StreamingSearchMode.java

@@ -11,38 +11,28 @@
 import org.opensearch.common.annotation.ExperimentalApi;
 
 /**
- * Defines the different streaming search strategies based on the design-by-case approach.
- * Each mode optimizes for different use cases and performance characteristics.
+ * Streaming search mode selector.
+ *
+ * <p>Only {@link #NO_SCORING} is active in the current streaming implementation.
+ * Other values are retained for request parsing/backward compatibility.
  *
  * @opensearch.internal
  */
 @ExperimentalApi
 public enum StreamingSearchMode {
 
     /**
-     * Case 1: No scoring, no sorting - fastest TTFB
-     * - Shard collector: StreamingUnsortedCollector
-     * - Ring buffer with batch emission
-     * - Round-robin merge at coordinator
-     * - Best for: simple filtering, counting, exists queries
+     * No scoring and no sorting.
      */
     NO_SCORING("no_scoring"),
 
     /**
-     * Case 2: Full scoring + explicit sort - production ready
-     * - Shard collector: StreamingSortedCollector
-     * - WAND/Block-Max WAND with windowed top-K heap
-     * - K-way streaming merge at coordinator
-     * - Best for: scored searches with sorting
+     * Retained for backward compatibility.
      */
     SCORED_SORTED("scored_sorted"),
 
     /**
-     * Case 3: Full scoring, no sorting - moderate performance
-     * - Shard collector: StreamingScoredUnsortedCollector
-     * - Ring buffer with scoring
-     * - No merge needed at coordinator
-     * - Best for: scored searches without sorting
+     * Retained for backward compatibility.
      */
     SCORED_UNSORTED("scored_unsorted");
 
@@ -68,7 +58,6 @@ public static StreamingSearchMode fromString(String mode) {
             return NO_SCORING; // Default
         }
 
-        // Backward compatibility: coerce older scored modes into NO_SCORING
         if ("SCORED_UNSORTED".equalsIgnoreCase(mode) || "SCORED_SORTED".equalsIgnoreCase(mode) || "NO_SCORING".equalsIgnoreCase(mode)) {
             return NO_SCORING;
         }
@@ -78,7 +67,7 @@ public static StreamingSearchMode fromString(String mode) {
                 return m;
             }
         }
-        
+
         throw new IllegalArgumentException("Unknown StreamingSearchMode: " + mode);
     }
 
@@ -98,8 +87,6 @@ public boolean requiresSorting() {
         return false; // No sorting is supported in the current mode
     }
 
-    // Confidence-based mode removed in this branch
-
     @Override
     public String toString() {
         return value;

server/src/main/java/org/opensearch/search/query/StreamingUnsortedCollectorContext.java

@@ -29,19 +29,6 @@
 /**
  * Streaming collector context for NO_SCORING mode.
  * Collects documents without scoring for fastest emission.
- *
- * Implements memory-bounded collection using a "firstK" pattern where only the first K
- * documents are retained for the final result. Documents are collected in batches
- * controlled by search.streaming.batch_size setting (default: 10, max: 100).
- *
- * Memory footprint: O(K + batchSize) where K is the requested number of hits.
- *
- * Circuit Breaker Policy:
- * - Batch buffers: No CB checks as they're strictly bounded (10-100 docs) and cleared after emission
- * - FirstK list: Protected by parent QueryPhaseResultConsumer's circuit breaker during final reduction
- * - Max memory per collector: ~8KB for batch (100 docs * 16 bytes) + ~80KB for firstK (10000 docs * 16 bytes)
- * - Decision rationale: The overhead of CB checks (atomic operations) would exceed the memory saved
- *   for such small, bounded allocations that are immediately released
  */
 public class StreamingUnsortedCollectorContext extends TopDocsCollectorContext {