xiangfu0 commented on code in PR #18649:
URL: https://github.com/apache/pinot/pull/18649#discussion_r3629151647
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/plan/server/ServerPlanRequestUtils.java:
##########
@@ -126,8 +128,12 @@ public static OpChain compileLeafStage(
}
int numRequests = instanceRequests.size();
List<ServerQueryRequest> serverQueryRequests = new
ArrayList<>(numRequests);
+ QueryExecutionContext queryExecutionContext =
QueryThreadContext.get().getExecutionContext();
for (InstanceRequest instanceRequest : instanceRequests) {
- serverQueryRequests.add(new ServerQueryRequest(instanceRequest,
ServerMetrics.get(), queryArrivalTimeMs, true));
+ ServerQueryRequest serverQueryRequest =
+ new ServerQueryRequest(instanceRequest, ServerMetrics.get(),
queryArrivalTimeMs, true);
+ serverQueryRequest.setExecutionContext(queryExecutionContext);
+ serverQueryRequests.add(serverQueryRequest);
Review Comment:
Fixed in 2b9cc00a24. ServerPlanRequestUtils now uses
QueryThreadContext.getIfAvailable() and only installs the execution context
when a query thread context is present, so direct/non-request execution paths
no longer fail.
##########
pinot-clients/pinot-cli/src/main/java/org/apache/pinot/cli/PinotCli.java:
##########
@@ -270,16 +284,22 @@ private void runSingle(Connection conn, String sql)
private void executeAndRender(Connection conn, String sql)
throws SQLException {
- String composed = prefixSessionOptions(sql);
+ String clientQueryId = "pinotcli" +
UUID.randomUUID().toString().replace("-", "");
+ String composed = prefixSessionOptions(sql, clientQueryId);
Instant start = Instant.now();
- Progress progress = new Progress();
+ boolean progressEnabled = _progressIntervalMs > 0 &&
isInteractiveProgressEnabled();
+ Progress progress = new Progress(getControllerProgressUrl(clientQueryId),
_headers, progressEnabled);
ScheduledExecutorService scheduler =
Executors.newSingleThreadScheduledExecutor();
- ScheduledFuture<?> spinner = scheduler.scheduleAtFixedRate(() ->
progress.tick(), 0, 120, TimeUnit.MILLISECONDS);
+ ScheduledFuture<?> spinner = null;
+ if (progressEnabled) {
+ spinner = scheduler.scheduleAtFixedRate(() -> progress.tick(), 0,
_progressIntervalMs, TimeUnit.MILLISECONDS);
+ }
Review Comment:
Fixed in 2b9cc00a24. The CLI now creates the progress object, scheduler, and
generated clientQueryId only when interactive progress is enabled; disabled and
redirected modes allocate no polling thread.
##########
pinot-controller/src/main/resources/app/pages/Query.tsx:
##########
@@ -371,15 +447,37 @@ const QueryPage = () => {
})
}
- const results = await PinotMethodUtils.getQueryResults(params);
- setResultError(results.exceptions || []);
- setResultData(results.result || { columns: [], records: [] });
- setQueryStats(results.queryStats || { columns: QUERY_STATS_COLUMNS,
records: [] });
- setOutputResult(JSON.stringify(results.data, null, 2) || '');
- setStageStats(results?.data?.stageStats || {});
- setWarnings(extractWarnings(results));
- setQueryLoader(false);
- queryExecuted.current = false;
+ let progressStopped = false;
+ let progressTimer = 0;
+ const pollQueryProgress = async () => {
+ try {
+ const response = await getClientQueryProgress(clientQueryId,
QUERY_PROGRESS_POLL_INTERVAL_MS);
+ setQueryProgress(response.data);
+ } catch (error) {
+ // The query might not be registered yet, or may already have
completed.
+ } finally {
+ if (!progressStopped) {
+ progressTimer = window.setTimeout(pollQueryProgress,
QUERY_PROGRESS_POLL_INTERVAL_MS);
+ }
+ }
+ };
+ progressTimer = window.setTimeout(pollQueryProgress,
QUERY_PROGRESS_POLL_INTERVAL_MS);
Review Comment:
Fixed in 2b9cc00a24. Query Console stores the polling timeout in a ref,
clears it in the request finally path and on unmount, and uses a generation
guard so stale callbacks cannot update a newer run.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/InstanceRequestHandler.java:
##########
@@ -266,6 +278,38 @@ public Set<String> getRunningQueryIds() {
return new HashSet<>(_executionContexts.keySet());
}
+ @Nullable
+ public QueryProgressStats getQueryProgressStats(String queryId) {
+ Preconditions.checkState(_executionContexts != null &&
_completedProgressStats != null,
+ "Query cancellation is not enabled on server");
+ QueryExecutionContext executionContext = _executionContexts.get(queryId);
+ return executionContext != null ? executionContext.getProgressStats()
+ : _completedProgressStats.getIfPresent(queryId);
+ }
+
+ private void retainCompletedProgressStats(String queryId,
QueryExecutionContext executionContext,
+ boolean successful) {
+ QueryProgressStats progressStats = executionContext.getProgressStats();
+ if (progressStats == null) {
+ return;
+ }
+ if (successful) {
+ progressStats = getCompletedProgressStats(progressStats);
+ }
+ _completedProgressStats.put(queryId, progressStats);
+ }
+
+ private static QueryProgressStats
getCompletedProgressStats(QueryProgressStats progressStats) {
+ long totalWorkUnits = progressStats.getTotalWorkUnits();
+ long processedWorkUnits = totalWorkUnits >= 0 ?
Math.max(progressStats.getProcessedWorkUnits(), totalWorkUnits)
+ : progressStats.getProcessedWorkUnits();
+ long totalSegmentsToProcess = progressStats.getTotalSegmentsToProcess();
+ long processedSegments = totalSegmentsToProcess >= 0
+ ? Math.max(progressStats.getProcessedSegments(),
totalSegmentsToProcess) : progressStats.getProcessedSegments();
+ return new QueryProgressStats(processedWorkUnits, totalWorkUnits,
processedSegments, totalSegmentsToProcess,
+ progressStats.isEstimated());
+ }
Review Comment:
Fixed in 2b9cc00a24. Successful completed snapshots clamp processed work
units to known totals before caching, preventing completed output such as 12/10
while preserving unknown totals.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
