feat: Added query mode for skipping incremental segments during query

docs/querying/query-context-reference.md

@@ -71,7 +71,7 @@ Unless otherwise noted, the following parameters apply to all query types, and t
 |`setProcessingThreadNames`|`false`| Flag indicating whether processing thread names will be set to `processing_<queryId>` while processing a query. Thread renaming aids in interpreting thread dumps, but has measurable thread renaming overhead when segment scans are very quick. |
 |`sqlPlannerBloat`|`1000`|Calcite parameter which controls whether to merge two Project operators when inlining expressions causes complexity to increase. Implemented as a workaround to exception `There are not enough rules to produce a node with desired properties: convention=DRUID, sort=[]` thrown after rejecting the merge of two projects.|
 |`cloneQueryMode`|`excludeClones`| Indicates whether clone Historicals should be queried by brokers. Clone servers are created by the `cloneServers` Coordinator dynamic configuration. Possible values are `excludeClones`, `includeClones` and `preferClones`. `excludeClones` means that clone Historicals are not queried by the broker. `preferClones` indicates that when given a choice between the clone Historical and the original Historical which is being cloned, the broker chooses the clones. Historicals which are not involved in the cloning process will still be queried. `includeClones` means that broker queries any Historical without regarding clone status. This parameter only affects native queries. MSQ does not query Historicals directly.|
-|`realtimeSegmentsMode` |`include`| Controls whether realtime segments are queried. `include` queries all segments, including realtime. `exclude` skips realtime segments. `exclusive` queries only realtime segments. |
+|`realtimeSegmentsMode` |`include`| Controls whether realtime segments are queried. `include` queries all segments, including realtime. `exclude` skips realtime segments. `exclude_incremental` skip only partial realtime segments. `exclusive` queries only realtime segments. |
 |`realtimeSegmentsOnly` |`false`| **Deprecated.** Use `realtimeSegmentsMode=exclusive` instead. When set to `true`, this is equivalent to `realtimeSegmentsMode=exclusive`. When set to `false`, this is equivalent to `realtimeSegmentsMode=include`.|
 
 ## Parameters by query type

processing/src/main/java/org/apache/druid/query/QueryContexts.java

@@ -256,7 +256,9 @@ public enum RealtimeSegmentsMode
     /** Query only realtime segments. */
     EXCLUSIVE,
     /** Skip realtime segments; query only historical. */
-    EXCLUDE;
+    EXCLUDE,
+    /** Skip partial in memory realtime segments that have not yet been indexed */
+    EXCLUDE_INCREMENTAL;
 
     @JsonCreator
     public static RealtimeSegmentsMode fromString(String str)

processing/src/test/java/org/apache/druid/query/QueryContextTest.java

@@ -458,6 +458,11 @@ public void testGetRealtimeSegmentsMode()
         QueryContext.of(ImmutableMap.of(QueryContexts.REALTIME_SEGMENTS_MODE, "exclude"))
                    .getRealtimeSegmentsMode()
     );
+    assertEquals(
+            QueryContexts.RealtimeSegmentsMode.EXCLUDE_INCREMENTAL,
+            QueryContext.of(ImmutableMap.of(QueryContexts.REALTIME_SEGMENTS_MODE, "exclude_incremental"))
+                    .getRealtimeSegmentsMode()
+    );
     assertEquals(
         QueryContexts.RealtimeSegmentsMode.INCLUDE,
         QueryContext.of(ImmutableMap.of(QueryContexts.REALTIME_SEGMENTS_MODE, "include"))
@@ -507,7 +512,7 @@ public void testGetRealtimeSegmentsModeInvalidValue()
                          .getRealtimeSegmentsMode()
     );
     assertEquals(
-        "Expected key [realtimeSegmentsMode] to be referring to one of the values [INCLUDE,EXCLUSIVE,EXCLUDE] of enum [RealtimeSegmentsMode], but got [badvalue]",
+        "Expected key [realtimeSegmentsMode] to be referring to one of the values [INCLUDE,EXCLUSIVE,EXCLUDE,EXCLUDE_INCREMENTAL] of enum [RealtimeSegmentsMode], but got [badvalue]",
         e.getMessage()
     );
   }

server/src/main/java/org/apache/druid/client/CachingClusteredClient.java

@@ -470,6 +470,7 @@ private Set<SegmentServerSelector> computeSegmentsToQuery(
                 continue;
               }
               break;
+            case EXCLUDE_INCREMENTAL:
             case INCLUDE:
               break;
           }

server/src/main/java/org/apache/druid/segment/realtime/appenderator/SinkQuerySegmentWalker.java

@@ -45,6 +45,7 @@
 import org.apache.druid.query.FinalizeResultsQueryRunner;
 import org.apache.druid.query.NoopQueryRunner;
 import org.apache.druid.query.Query;
+import org.apache.druid.query.QueryContexts;
 import org.apache.druid.query.QueryDataSource;
 import org.apache.druid.query.QueryMetrics;
 import org.apache.druid.query.QueryPlus;
@@ -186,7 +187,9 @@ public <T> QueryRunner<T> getQueryRunnerForSegments(final Query<T> query, final
     }
 
     final QueryToolChest<T, Query<T>> toolChest = factory.getToolchest();
-    final boolean skipIncrementalSegment = query.context().getBoolean(CONTEXT_SKIP_INCREMENTAL_SEGMENT, false);
+    // Supporting undocumented CONTEXT_SKIP_INCREMENTAL_SEGMENT flag for backward compatibility
+    final boolean skipIncrementalSegment = (query.context().getRealtimeSegmentsMode() == QueryContexts.RealtimeSegmentsMode.EXCLUDE_INCREMENTAL)
+                                              || query.context().getBoolean(CONTEXT_SKIP_INCREMENTAL_SEGMENT, false);
     final AtomicLong cpuTimeAccumulator = new AtomicLong(0L);
 
     // Make sure this query type can handle the subquery, if present.