More cleanup

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

Author: atris

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

@@ -848,29 +848,6 @@ QueryPhaseResultConsumer newSearchPhaseResults(
         }
     }
 
-    /**
-     * Returns a new {@link StreamQueryPhaseResultConsumer} instance that reduces search responses incrementally.
-     */
-    StreamQueryPhaseResultConsumer newStreamSearchPhaseResults(
-        Executor executor,
-        CircuitBreaker circuitBreaker,
-        SearchProgressListener listener,
-        SearchRequest request,
-        int numShards,
-        Consumer<Exception> onPartialMergeFailure
-    ) {
-        return new StreamQueryPhaseResultConsumer(
-            request,
-            executor,
-            circuitBreaker,
-            this,
-            listener,
-            namedWriteableRegistry,
-            numShards,
-            onPartialMergeFailure,
-            null // No ClusterSettings in this legacy path
-        );
-    }
 
     /**
      * The top docs statistics

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

@@ -26,6 +26,19 @@
 /**
  * Query phase result consumer for streaming search.
  * Supports progressive batch reduction with configurable scoring modes.
+ * 
+ * Batch reduction frequency is controlled by per-mode multipliers from cluster settings:
+ * - NO_SCORING: Immediate reduction (batch size = 1) for fastest time-to-first-byte
+ * - SCORED_UNSORTED: Small batches controlled by search.streaming.scored_unsorted.batch_multiplier (default: 2)
+ * - CONFIDENCE_BASED: Moderate batches controlled by search.streaming.confidence.batch_multiplier (default: 3)
+ * - SCORED_SORTED: Larger batches controlled by search.streaming.scored_sorted.batch_multiplier (default: 10)
+ * 
+ * These multipliers are applied to the base batch reduce size (typically 5) to determine
+ * how many shard results are accumulated before triggering a partial reduction. Lower values
+ * mean more frequent reductions and faster streaming, but higher coordinator CPU usage.
+ * 
+ * ClusterSettings must be provided (non-null) to enable dynamic configuration. Tests should
+ * provide a properly configured ClusterSettings instance rather than null.
  *
  * @opensearch.internal
  */
@@ -37,6 +50,11 @@ public class StreamQueryPhaseResultConsumer extends QueryPhaseResultConsumer {
     private final ClusterSettings clusterSettings;
     private int resultsReceived = 0;
 
+    /**
+     * Creates a streaming query phase result consumer.
+     * 
+     * @param clusterSettings cluster settings for dynamic multipliers (must not be null)
+     */
     public StreamQueryPhaseResultConsumer(
         SearchRequest request,
         Executor executor,
@@ -62,6 +80,11 @@ public StreamQueryPhaseResultConsumer(
         // Initialize scoring mode from request
         String mode = request.getStreamingSearchMode();
         this.scoringMode = (mode != null) ? StreamingSearchMode.fromString(mode) : StreamingSearchMode.SCORED_SORTED;
+        
+        // ClusterSettings is required for dynamic configuration
+        if (clusterSettings == null) {
+            throw new IllegalArgumentException("ClusterSettings must not be null for StreamQueryPhaseResultConsumer");
+        }
         this.clusterSettings = clusterSettings;
     }
 
@@ -74,7 +97,8 @@ 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) {
+        // In this case, clusterSettings is also null, so use a sensible default
+        if (scoringMode == null || clusterSettings == null) {
             return super.getBatchReduceSize(requestBatchedReduceSize, minBatchReduceSize * 10);
         }
 
@@ -84,26 +108,18 @@ int getBatchReduceSize(int requestBatchedReduceSize, int minBatchReduceSize) {
                 return Math.min(requestBatchedReduceSize, 1);
             case SCORED_UNSORTED:
                 // Small batches for quick emission without sorting overhead
-                int suMult = clusterSettings != null
-                    ? clusterSettings.get(StreamingSearchSettings.STREAMING_SCORED_UNSORTED_BATCH_MULTIPLIER)
-                    : 2;
+                int suMult = clusterSettings.get(StreamingSearchSettings.STREAMING_SCORED_UNSORTED_BATCH_MULTIPLIER);
                 return super.getBatchReduceSize(requestBatchedReduceSize, minBatchReduceSize * suMult);
             case CONFIDENCE_BASED:
                 // Moderate batching for progressive emission with confidence
-                int cMult = clusterSettings != null
-                    ? clusterSettings.get(StreamingSearchSettings.STREAMING_CONFIDENCE_BATCH_MULTIPLIER)
-                    : 3;
+                int cMult = clusterSettings.get(StreamingSearchSettings.STREAMING_CONFIDENCE_BATCH_MULTIPLIER);
                 return super.getBatchReduceSize(requestBatchedReduceSize, minBatchReduceSize * cMult);
             case SCORED_SORTED:
                 // Higher batch size to collect more results before reducing (sorting is expensive)
-                int ssMult = clusterSettings != null
-                    ? clusterSettings.get(StreamingSearchSettings.STREAMING_SCORED_SORTED_BATCH_MULTIPLIER)
-                    : 10;
+                int ssMult = clusterSettings.get(StreamingSearchSettings.STREAMING_SCORED_SORTED_BATCH_MULTIPLIER);
                 return super.getBatchReduceSize(requestBatchedReduceSize, minBatchReduceSize * ssMult);
             default:
-                int defMult = clusterSettings != null
-                    ? clusterSettings.get(StreamingSearchSettings.STREAMING_SCORED_SORTED_BATCH_MULTIPLIER)
-                    : 10;
+                int defMult = clusterSettings.get(StreamingSearchSettings.STREAMING_SCORED_SORTED_BATCH_MULTIPLIER);
                 return super.getBatchReduceSize(requestBatchedReduceSize, minBatchReduceSize * defMult);
         }
     }

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

@@ -1312,9 +1312,7 @@ AbstractSearchAsyncAction<? extends SearchPhaseResult> searchAsyncAction(
                     );
                     break;
                 case QUERY_THEN_FETCH:
-                    System.out.println("DEBUG: isStreamingRequest=" + isStreamingRequest + ", streamSearchTransportService=" + (streamSearchTransportService != null));
                     if (isStreamingRequest && streamSearchTransportService != null) {
-                        System.out.println("DEBUG: Using StreamSearchQueryThenFetchAsyncAction!");
                         searchAsyncAction = new StreamSearchQueryThenFetchAsyncAction(
                             logger,
                             streamSearchTransportService,

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

@@ -584,9 +584,25 @@ public boolean isStreamingSearch() {
     }
 
     /**
-     * Returns the configured batch size for streaming emissions.
-     * Default implementation returns 10. Implementations may override to pull dynamic settings.
-     * @return the batch size for streaming collection
+     * Returns the configured batch size for streaming document collection.
+     * 
+     * This value controls how many documents are collected in memory before emission
+     * during streaming search operations. It reads the dynamic cluster setting
+     * 'search.streaming.batch_size' (via StreamingSearchSettings.STREAMING_BATCH_SIZE)
+     * when ClusterSettings is available, with a default of 10 and max of 100.
+     * 
+     * Default implementation returns 10. Concrete implementations (DefaultSearchContext)
+     * override this to read from ClusterSettings when available, falling back to 10
+     * if ClusterSettings is not injected or the setting is not configured.
+     * 
+     * The batch size affects streaming performance:
+     * - Smaller values (1-10): Lower latency, more frequent emissions
+     * - Larger values (50-100): Higher throughput, less network overhead
+     * 
+     * Note: This value is read once per search context creation and does not update
+     * dynamically during a search operation.
+     * 
+     * @return the batch size for streaming collection (default: 10, range: 1-100)
      */
     public int getStreamingBatchSize() {
         return 10;

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

@@ -23,7 +23,21 @@
 import java.util.concurrent.atomic.AtomicLong;
 
 /**
- * Collector context for confidence-based streaming mode.
+ * Streaming collector context for CONFIDENCE_BASED mode.
+ * Collects documents with scores for progressive confidence-based emission.
+ * 
+ * Implements memory-bounded collection using a "topK" pattern where the best K
+ * documents by score are retained. 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  
+ * - TopK list: Protected by parent QueryPhaseResultConsumer's circuit breaker during final reduction
+ * - Max memory per collector: ~8KB for batch (100 docs * 16 bytes) + ~80KB for topK (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 StreamingConfidenceCollectorContext extends TopDocsCollectorContext {
 

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

@@ -21,7 +21,21 @@
 import java.util.concurrent.atomic.AtomicLong;
 
 /**
- * Collector context for scored unsorted streaming mode.
+ * Streaming collector context for SCORED_UNSORTED mode.
+ * Collects documents with scores but without sorting for fast emission with relevance.
+ * 
+ * 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 StreamingScoredUnsortedCollectorContext extends TopDocsCollectorContext {
 

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

@@ -18,7 +18,22 @@
 import java.util.List;
 
 /**
- * Collector context for sorted streaming mode.
+ * Streaming collector context for SCORED_SORTED mode.
+ * Collects and maintains documents in sorted order (by score or custom sort).
+ * 
+ * Uses Lucene's TopScoreDocCollectorManager for efficient sorted collection with
+ * incremental merging. Documents are collected in larger batches (10x default multiplier)
+ * to amortize sorting costs, controlled by search.streaming.scored_sorted.batch_multiplier.
+ * 
+ * Memory footprint: O(K) where K is the requested number of hits.
+ * The TopScoreDocCollector maintains a min-heap of size K.
+ * 
+ * Circuit Breaker Policy:
+ * - Heap structure: Protected by TopScoreDocCollector's internal memory management
+ * - Parent reduction: Protected by QueryPhaseResultConsumer's circuit breaker
+ * - Max memory per collector: ~80KB for topK heap (10000 docs * 16 bytes)
+ * - Decision rationale: Sorting requires maintaining all K docs in memory, but Lucene's
+ *   collectors are already optimized for memory efficiency
  */
 public class StreamingSortedCollectorContext extends TopDocsCollectorContext {
 

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

@@ -25,6 +25,19 @@
 /**
  * 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 {