klsince commented on code in PR #9141:
URL: https://github.com/apache/pinot/pull/9141#discussion_r934782007
##########
pinot-core/src/main/java/org/apache/pinot/core/util/trace/TraceContext.java:
##########
@@ -45,9 +49,28 @@
* request from tracing to prevent resource leak.
*/
public final class TraceContext {
+
+ static Map<Long, Pair<QueryContext, Future[]>> requestId2FuturesMap = new
ConcurrentHashMap<>();
+
private TraceContext() {
}
+ public static void register(long requestId, QueryContext queryContext,
Future[] futures) {
+ requestId2FuturesMap.put(requestId, Pair.of(queryContext, futures));
+ }
+
+ public static void cancel(long requestId) {
+ Pair<QueryContext, Future[]> queryContextFuturesPair =
requestId2FuturesMap.get(requestId);
Review Comment:
use remove() instead? to avoid keeping futures around.
alternatively, can also clean things up in unregister() method.
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/combine/BaseCombineOperator.java:
##########
@@ -73,6 +74,13 @@ protected BaseCombineOperator(List<Operator> operators,
QueryContext queryContex
// The parallelism is bounded by the task count.
_numTasks = CombineOperatorUtils.getNumTasksForQuery(operators.size(),
queryContext.getMaxExecutionThreads());
_futures = new Future[_numTasks];
+
+ try {
+ long requestId =
Long.parseLong(queryContext.getQueryOptions().get("requestId"));
+ TraceContext.register(requestId, queryContext, _futures);
Review Comment:
Would suggest not to reuse `TraceContext` and keep it just for
tracing/monitoring purpose. Perhaps create sth similar to control query at
runtime, e.g. `QueryRuntimeContext` to track the futures.
But in general, I'd prefer add cancel() on Operator and Plan interface to
make it explicit that query is cancellable. The reason to add cancel() on Plan
interface is because `queryPlan.execute()` is the entry point of query
execution, so we can have `queryPlan.cancel()` to cancel a query.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java:
##########
@@ -1472,6 +1490,9 @@ static void setOptions(PinotQuery pinotQuery, long
requestId, String query, Json
if (enableTrace) {
queryOptions.put(Broker.Request.TRACE, "true");
}
+
+ queryOptions.put(CommonConstants.Query.Request.MetadataKeys.REQUEST_ID,
String.valueOf(requestId));
Review Comment:
I assume this is just to pass `requestId` to servers? The `requestId` is
already passed over, and can be read via `queryRequest.getRequestId()` at
server side
e.g.
```
ServerQueryExecutorV1Impl.processQuery() {
...
try {
Tracing.getTracer().register(queryRequest.getRequestId());
...
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
