This is an automated email from the ASF dual-hosted git repository.
abhishekrb pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 96bd4ebf70b Add optional realtimeSegmentsOnly in query context (#18329)
96bd4ebf70b is described below
commit 96bd4ebf70b9cbac81efc96ce653b63bdced6888
Author: aho135 <[email protected]>
AuthorDate: Thu Aug 28 07:16:07 2025 -0700
Add optional realtimeSegmentsOnly in query context (#18329)
Adds an optional query context called realtimeSegmentsOnly that when set to
true will only return results from realtime segments.
---------
Co-authored-by: Andrew Ho <[email protected]>
---
docs/querying/query-context-reference.md | 1 +
.../java/org/apache/druid/query/QueryContext.java | 5 +++
.../java/org/apache/druid/query/QueryContexts.java | 3 ++
.../org/apache/druid/query/QueryContextTest.java | 11 +++++
.../druid/client/CachingClusteredClient.java | 4 ++
.../druid/client/CachingClusteredClientTest.java | 50 ++++++++++++++++++++++
.../query-context-completions.ts | 4 ++
7 files changed, 78 insertions(+)
diff --git a/docs/querying/query-context-reference.md
b/docs/querying/query-context-reference.md
index acc762fa9fe..52515236620 100644
--- a/docs/querying/query-context-reference.md
+++ b/docs/querying/query-context-reference.md
@@ -70,6 +70,7 @@ Unless otherwise noted, the following parameters apply to all
query types, and t
|`setProcessingThreadNames`|`true`| Whether processing thread names will be
set to `queryType_dataSource_intervals` while processing a query. This aids in
interpreting thread dumps, and is on by default. Query overhead can be reduced
slightly by setting this to `false`. This has a tiny effect in most scenarios,
but can be meaningful in high-QPS, low-per-segment-processing-time scenarios. |
|`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 [...]
+|`realtimeSegmentsOnly` |`false`| When set to true, only query realtime
segments. Historical segments are excluded. |
## Parameters by query type
diff --git a/processing/src/main/java/org/apache/druid/query/QueryContext.java
b/processing/src/main/java/org/apache/druid/query/QueryContext.java
index b90be50e89a..096fd4afe9b 100644
--- a/processing/src/main/java/org/apache/druid/query/QueryContext.java
+++ b/processing/src/main/java/org/apache/druid/query/QueryContext.java
@@ -747,4 +747,9 @@ public class QueryContext
{
return getBoolean(QueryContexts.CTX_PREPLANNED,
QueryContexts.DEFAULT_PREPLANNED);
}
+
+ public boolean isRealtimeSegmentsOnly()
+ {
+ return getBoolean(QueryContexts.REALTIME_SEGMENTS_ONLY,
QueryContexts.DEFAULT_REALTIME_SEGMENTS_ONLY);
+ }
}
diff --git a/processing/src/main/java/org/apache/druid/query/QueryContexts.java
b/processing/src/main/java/org/apache/druid/query/QueryContexts.java
index d288775977c..0619e521b7a 100644
--- a/processing/src/main/java/org/apache/druid/query/QueryContexts.java
+++ b/processing/src/main/java/org/apache/druid/query/QueryContexts.java
@@ -137,6 +137,9 @@ public class QueryContexts
public static final String NATIVE_QUERY_SQL_PLANNING_MODE_COUPLED =
"COUPLED";
public static final String NATIVE_QUERY_SQL_PLANNING_MODE_DECOUPLED =
"DECOUPLED";
+ public static final String REALTIME_SEGMENTS_ONLY = "realtimeSegmentsOnly";
+ public static final boolean DEFAULT_REALTIME_SEGMENTS_ONLY = false;
+
public static final String CTX_PREPLANNED = "prePlanned";
public static final boolean DEFAULT_PREPLANNED = true;
diff --git
a/processing/src/test/java/org/apache/druid/query/QueryContextTest.java
b/processing/src/test/java/org/apache/druid/query/QueryContextTest.java
index 277713abe33..0cb931e50d2 100644
--- a/processing/src/test/java/org/apache/druid/query/QueryContextTest.java
+++ b/processing/src/test/java/org/apache/druid/query/QueryContextTest.java
@@ -429,6 +429,17 @@ public class QueryContextTest
assertNotNull(timeseries.getContext());
}
+ @Test
+ public void testIsRealtimeSegmentsOnly()
+ {
+ assertFalse(QueryContext.empty().isRealtimeSegmentsOnly());
+ assertTrue(
+ QueryContext
+ .of(ImmutableMap.of(QueryContexts.REALTIME_SEGMENTS_ONLY, true))
+ .isRealtimeSegmentsOnly()
+ );
+ }
+
@Test
public void testSerialization() throws Exception
{
diff --git
a/server/src/main/java/org/apache/druid/client/CachingClusteredClient.java
b/server/src/main/java/org/apache/druid/client/CachingClusteredClient.java
index e0c7f0e4e5f..33237f372a2 100644
--- a/server/src/main/java/org/apache/druid/client/CachingClusteredClient.java
+++ b/server/src/main/java/org/apache/druid/client/CachingClusteredClient.java
@@ -457,6 +457,7 @@ public class CachingClusteredClient implements
QuerySegmentWalker
filterFieldsForPruning = null;
}
+ boolean isRealtimeSegmentOnly = query.context().isRealtimeSegmentsOnly();
// Filter unneeded chunks based on partition dimension
for (TimelineObjectHolder<String, ServerSelector> holder :
serversLookup) {
final Set<PartitionChunk<ServerSelector>> filteredChunks;
@@ -473,6 +474,9 @@ public class CachingClusteredClient implements
QuerySegmentWalker
}
for (PartitionChunk<ServerSelector> chunk : filteredChunks) {
ServerSelector server = chunk.getObject();
+ if (isRealtimeSegmentOnly && !server.isRealtimeSegment()) {
+ continue; // Skip historical segments when only realtime segments
are requested
+ }
final SegmentDescriptor segment = new SegmentDescriptor(
holder.getInterval(),
holder.getVersion(),
diff --git
a/server/src/test/java/org/apache/druid/client/CachingClusteredClientTest.java
b/server/src/test/java/org/apache/druid/client/CachingClusteredClientTest.java
index 1318a7c344a..621bbf0a044 100644
---
a/server/src/test/java/org/apache/druid/client/CachingClusteredClientTest.java
+++
b/server/src/test/java/org/apache/druid/client/CachingClusteredClientTest.java
@@ -3099,6 +3099,56 @@ public class CachingClusteredClientTest
Assert.assertNotEquals(etag1, etag2);
}
+ @Test
+ public void testRealtimeSegmentsQueryContext()
+ {
+ final Interval interval = Intervals.of("2016-01-01/2016-01-02");
+ final Interval queryInterval =
Intervals.of("2016-01-01T14:00:00/2016-01-02T14:00:00");
+ final DataSegment dataSegment = new DataSegment(
+ "dataSource",
+ interval,
+ "ver",
+ ImmutableMap.of(
+ "type", "hdfs",
+ "path", "/tmp"
+ ),
+ ImmutableList.of("product"),
+ ImmutableList.of("visited_sum"),
+ NoneShardSpec.instance(),
+ 9,
+ 12334
+ );
+ final ServerSelector selector = new ServerSelector(
+ dataSegment,
+ new HighestPriorityTierSelectorStrategy(new
RandomServerSelectorStrategy()),
+ HistoricalFilter.IDENTITY_FILTER
+ );
+ selector.addServerAndUpdateSegment(new QueryableDruidServer(servers[0],
null), dataSegment);
+ timeline.add(interval, "ver", new SingleElementPartitionChunk<>(selector));
+
+ final TimeBoundaryQuery query = Druids.newTimeBoundaryQueryBuilder()
+ .dataSource(DATA_SOURCE)
+ .intervals(new
MultipleIntervalSegmentSpec(ImmutableList.of(queryInterval)))
+ .context(ImmutableMap.of("realtimeSegmentsOnly", false))
+ .randomQueryId()
+ .build();
+
+ final TimeBoundaryQuery query2 = Druids.newTimeBoundaryQueryBuilder()
+ .dataSource(DATA_SOURCE)
+ .intervals(new
MultipleIntervalSegmentSpec(ImmutableList.of(queryInterval)))
+ .context(ImmutableMap.of("realtimeSegmentsOnly", true))
+ .randomQueryId()
+ .build();
+
+ final ResponseContext responseContext = initializeResponseContext();
+
+ getDefaultQueryRunner().run(QueryPlus.wrap(query), responseContext);
+ getDefaultQueryRunner().run(QueryPlus.wrap(query2), responseContext);
+ final Map<String, Integer> remainingResponseMap = (Map<String, Integer>)
responseContext.get(ResponseContext.Keys.REMAINING_RESPONSES_FROM_QUERY_SERVERS);
+ Assert.assertEquals(1, remainingResponseMap.get(query.getId()).intValue());
+ Assert.assertEquals(0,
remainingResponseMap.get(query2.getId()).intValue());
+ }
+
@SuppressWarnings("unchecked")
private QueryRunner getDefaultQueryRunner()
{
diff --git
a/web-console/src/dialogs/edit-context-dialog/query-context-completions.ts
b/web-console/src/dialogs/edit-context-dialog/query-context-completions.ts
index 0b7abe19e7c..e74513c7e28 100644
--- a/web-console/src/dialogs/edit-context-dialog/query-context-completions.ts
+++ b/web-console/src/dialogs/edit-context-dialog/query-context-completions.ts
@@ -72,6 +72,10 @@ export const QUERY_CONTEXT_COMPLETIONS: JsonCompletionRule[]
= [
value: 'vectorize',
documentation: 'Enable vectorized query execution',
},
+ {
+ value: 'realtimeSegmentsOnly',
+ documentation: 'Whether to query only realtime segments',
+ },
{
value: 'vectorSize',
documentation: 'Vector size for vectorized execution',
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
