codeant-ai-for-open-source[bot] commented on code in PR #43424:
URL: https://github.com/apache/superset/pull/43424#discussion_r3836572566
##########
superset-frontend/src/middleware/asyncEvent.ts:
##########
@@ -17,326 +17,207 @@
* under the License.
*/
import {
- ensureIsArray,
isFeatureEnabled,
FeatureFlag,
makeApi,
SupersetClient,
- getClientErrorObject,
- parseErrorJson,
- SupersetError,
} from '@superset-ui/core';
import { logging } from '@apache-superset/core/utils';
import getBootstrapData from 'src/utils/getBootstrapData';
-type AsyncEvent = {
- id?: string | null;
- channel_id: string;
- job_id: string;
- user_id?: string;
- status: string;
- errors?: SupersetError[];
- result_url: string | null;
+// The GTF task type chart-data queries run under (see
+// superset/tasks/async_queries.py CHART_QUERY_TASK). Polling is filtered to
this
+// type so a dashboard only tracks its own chart-data work, not every task.
+const CHART_QUERY_TASK_TYPE = 'superset.query_object_v1';
+const STATUS_CHANGES_URL = '/api/v1/task/status_changes';
+
+// Terminal GTF task statuses (mirror superset_core.tasks.types.TaskStatus).
+const STATUS_SUCCESS = 'success';
+const TERMINAL_STATUSES = new Set([
+ STATUS_SUCCESS,
+ 'failure',
+ 'aborted',
+ 'timed_out',
+]);
+
+type TaskStatusChange = { status: string; progress: number | null };
+type StatusChangesResponse = {
+ statuses: Record<string, TaskStatusChange>;
+ cursor: string | null;
};
-type CachedDataResponse = {
- status: string;
- data: any;
-};
+// The 202 body from POST /chart/data when async: the query tasks to await.
+export type AsyncJob = { task_ids: string[] };
+
type AppConfig = Record<string, any>;
-type ListenerFn = (asyncEvent: AsyncEvent) => Promise<any>;
-const TRANSPORT_POLLING = 'polling';
-const TRANSPORT_WS = 'ws';
-const JOB_STATUS = {
- PENDING: 'pending',
- RUNNING: 'running',
- ERROR: 'error',
- DONE: 'done',
+type Waiter = {
+ pending: Set<string>;
+ failed: boolean;
+ // Re-issue the original chart-data request once every task has succeeded;
the
+ // per-query DATA cache is now warm, so it returns synchronously (200).
+ resolve: () => void;
+ reject: (error: unknown) => void;
+ signal?: AbortSignal;
+ onAbort?: () => void;
};
-const LOCALSTORAGE_KEY = 'last_async_event_id';
-const POLLING_URL = '/api/v1/async_event/';
-const MAX_RETRIES = 6;
-const RETRY_DELAY = 100;
-// Cap for the exponential backoff applied when polling requests fail
-// repeatedly (e.g. expired session, server or network errors)
-const MAX_ERROR_POLLING_DELAY_MS = 60000;
let config: AppConfig;
-let transport: string;
let pollingDelayMs: number;
let pollingTimeoutId: number;
-let listenersByJobId: Map<string, ListenerFn>;
-let retriesByJobId: Map<string, number>;
-let lastReceivedEventId: string | null | undefined;
-let consecutivePollingErrorCount = 0;
-// Incremented on every init() so polling invocations that are already
-// awaiting a fetch when re-init happens can detect they are stale and
-// stop, instead of mutating fresh state or scheduling a second loop
+// Registry of in-flight waiters keyed by every task uuid they await, so a
single
+// shared poll loop fans status changes out to whichever request is awaiting
them.
+let waitersByTaskId: Map<string, Waiter>;
+// Server-issued watermark: fetched as a baseline at init (before any chart
query
+// is triggered) so no task created afterwards is missed, then advanced by each
+// poll. Always the server's own clock, never the browser's.
+let cursor: string | null;
+let baselineReady: Promise<void> | null;
+// Incremented on every init() so an in-flight poll can detect it is stale and
+// stop instead of scheduling a second loop or mutating fresh state.
let pollingGeneration = 0;
-const addListener = (id: string, fn: ListenerFn) => {
- listenersByJobId.set(id, fn);
-};
-
-const removeListener = (id: string) => {
- if (!listenersByJobId.has(id)) return;
- listenersByJobId.delete(id);
-};
-
-const fetchCachedData = async (
- asyncEvent: AsyncEvent,
- signal?: AbortSignal,
-): Promise<CachedDataResponse> => {
- let status = 'success';
- let data;
- try {
- const { json } = await SupersetClient.get({
- endpoint: String(asyncEvent.result_url),
- signal,
- });
- data = 'result' in json ? json.result : json;
- } catch (response) {
- status = 'error';
- data = await getClientErrorObject(response);
- }
-
- return { status, data };
-};
+const fetchStatusChanges = makeApi<
+ { cursor?: string | null; task_type: string },
+ StatusChangesResponse
+>({
+ method: 'GET',
+ endpoint: STATUS_CHANGES_URL,
+});
-const cancelAsyncJob = (jobId: string) => {
- // Best-effort server-side cancel; the request stops the running Celery task
- // so it no longer consumes warehouse resources. Failures are non-fatal: the
- // client has already stopped waiting on the job.
+const cancelTask = (taskId: string) => {
+ // Best-effort server-side cancel so an abandoned query stops consuming
+ // warehouse resources. Failures are non-fatal: the client has already
stopped
+ // waiting on the task.
SupersetClient.post({
- endpoint: `/api/v1/async_event/${jobId}/cancel`,
+ endpoint: `/api/v1/task/${taskId}/cancel`,
}).catch(error => {
- logging.warn('Failed to cancel async job', jobId, error);
+ logging.warn('Failed to cancel task', taskId, error);
});
Review Comment:
**Suggestion:** Aborting an embedded guest request calls the generic task
cancellation endpoint, but guest authorization is based on `guest_key` while
the cancellation command validates only `get_user_id()` and
`has_subscriber(user_id)`. The cancellation therefore fails for guest-created
shared tasks, leaving the worker running despite the client abandoning the
request. Guest cancellation must use the guest subscriber identity or avoid
claiming cancellation succeeded. [api mismatch]
<details>
<summary><b>Severity Level:</b> Major ⚠️</summary>
```mdx
- ⚠️ Embedded guest cancellations leave warehouse queries running.
- ⚠️ Abandoned guest requests continue consuming GTF capacity.
- ❌ Guest cancellation cannot unsubscribe or abort shared tasks.
```
</details>
[](https://docs.codeant.ai/cli/resolve-pr-comments-skill)
<details>
<summary><b>Prompt for AI Agent 🤖 </b></summary>
```mdx
This is a comment left during a code review.
**Path:** superset-frontend/src/middleware/asyncEvent.ts
**Line:** 92:96
**Comment:**
*Api Mismatch: Aborting an embedded guest request calls the generic
task cancellation endpoint, but guest authorization is based on `guest_key`
while the cancellation command validates only `get_user_id()` and
`has_subscriber(user_id)`. The cancellation therefore fails for guest-created
shared tasks, leaving the worker running despite the client abandoning the
request. Guest cancellation must use the guest subscriber identity or avoid
claiming cancellation succeeded.
Validate the correctness of the flagged issue. If correct, How can I resolve
this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask
user if the user wants to fix the rest of the comments as well. if said yes,
then fetch all the comments validate the correctness and implement a minimal fix
```
</details>
<a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F43424&comment_hash=cb8f9b76a72b9268e259aab65cb4da60d1062fe1372d4aeb4f0a19b8595f8201&reaction=like'>👍</a>
| <a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F43424&comment_hash=cb8f9b76a72b9268e259aab65cb4da60d1062fe1372d4aeb4f0a19b8595f8201&reaction=dislike'>👎</a>
##########
superset-frontend/src/middleware/asyncEvent.ts:
##########
@@ -17,326 +17,207 @@
* under the License.
*/
import {
- ensureIsArray,
isFeatureEnabled,
FeatureFlag,
makeApi,
SupersetClient,
- getClientErrorObject,
- parseErrorJson,
- SupersetError,
} from '@superset-ui/core';
import { logging } from '@apache-superset/core/utils';
import getBootstrapData from 'src/utils/getBootstrapData';
-type AsyncEvent = {
- id?: string | null;
- channel_id: string;
- job_id: string;
- user_id?: string;
- status: string;
- errors?: SupersetError[];
- result_url: string | null;
+// The GTF task type chart-data queries run under (see
+// superset/tasks/async_queries.py CHART_QUERY_TASK). Polling is filtered to
this
+// type so a dashboard only tracks its own chart-data work, not every task.
+const CHART_QUERY_TASK_TYPE = 'superset.query_object_v1';
+const STATUS_CHANGES_URL = '/api/v1/task/status_changes';
+
+// Terminal GTF task statuses (mirror superset_core.tasks.types.TaskStatus).
+const STATUS_SUCCESS = 'success';
+const TERMINAL_STATUSES = new Set([
+ STATUS_SUCCESS,
+ 'failure',
+ 'aborted',
+ 'timed_out',
+]);
+
+type TaskStatusChange = { status: string; progress: number | null };
+type StatusChangesResponse = {
+ statuses: Record<string, TaskStatusChange>;
+ cursor: string | null;
};
-type CachedDataResponse = {
- status: string;
- data: any;
-};
+// The 202 body from POST /chart/data when async: the query tasks to await.
+export type AsyncJob = { task_ids: string[] };
+
type AppConfig = Record<string, any>;
-type ListenerFn = (asyncEvent: AsyncEvent) => Promise<any>;
-const TRANSPORT_POLLING = 'polling';
-const TRANSPORT_WS = 'ws';
-const JOB_STATUS = {
- PENDING: 'pending',
- RUNNING: 'running',
- ERROR: 'error',
- DONE: 'done',
+type Waiter = {
+ pending: Set<string>;
+ failed: boolean;
+ // Re-issue the original chart-data request once every task has succeeded;
the
+ // per-query DATA cache is now warm, so it returns synchronously (200).
+ resolve: () => void;
+ reject: (error: unknown) => void;
+ signal?: AbortSignal;
+ onAbort?: () => void;
};
-const LOCALSTORAGE_KEY = 'last_async_event_id';
-const POLLING_URL = '/api/v1/async_event/';
-const MAX_RETRIES = 6;
-const RETRY_DELAY = 100;
-// Cap for the exponential backoff applied when polling requests fail
-// repeatedly (e.g. expired session, server or network errors)
-const MAX_ERROR_POLLING_DELAY_MS = 60000;
let config: AppConfig;
-let transport: string;
let pollingDelayMs: number;
let pollingTimeoutId: number;
-let listenersByJobId: Map<string, ListenerFn>;
-let retriesByJobId: Map<string, number>;
-let lastReceivedEventId: string | null | undefined;
-let consecutivePollingErrorCount = 0;
-// Incremented on every init() so polling invocations that are already
-// awaiting a fetch when re-init happens can detect they are stale and
-// stop, instead of mutating fresh state or scheduling a second loop
+// Registry of in-flight waiters keyed by every task uuid they await, so a
single
+// shared poll loop fans status changes out to whichever request is awaiting
them.
+let waitersByTaskId: Map<string, Waiter>;
+// Server-issued watermark: fetched as a baseline at init (before any chart
query
+// is triggered) so no task created afterwards is missed, then advanced by each
+// poll. Always the server's own clock, never the browser's.
+let cursor: string | null;
+let baselineReady: Promise<void> | null;
+// Incremented on every init() so an in-flight poll can detect it is stale and
+// stop instead of scheduling a second loop or mutating fresh state.
let pollingGeneration = 0;
-const addListener = (id: string, fn: ListenerFn) => {
- listenersByJobId.set(id, fn);
-};
-
-const removeListener = (id: string) => {
- if (!listenersByJobId.has(id)) return;
- listenersByJobId.delete(id);
-};
-
-const fetchCachedData = async (
- asyncEvent: AsyncEvent,
- signal?: AbortSignal,
-): Promise<CachedDataResponse> => {
- let status = 'success';
- let data;
- try {
- const { json } = await SupersetClient.get({
- endpoint: String(asyncEvent.result_url),
- signal,
- });
- data = 'result' in json ? json.result : json;
- } catch (response) {
- status = 'error';
- data = await getClientErrorObject(response);
- }
-
- return { status, data };
-};
+const fetchStatusChanges = makeApi<
+ { cursor?: string | null; task_type: string },
+ StatusChangesResponse
+>({
+ method: 'GET',
+ endpoint: STATUS_CHANGES_URL,
+});
-const cancelAsyncJob = (jobId: string) => {
- // Best-effort server-side cancel; the request stops the running Celery task
- // so it no longer consumes warehouse resources. Failures are non-fatal: the
- // client has already stopped waiting on the job.
+const cancelTask = (taskId: string) => {
+ // Best-effort server-side cancel so an abandoned query stops consuming
+ // warehouse resources. Failures are non-fatal: the client has already
stopped
+ // waiting on the task.
SupersetClient.post({
- endpoint: `/api/v1/async_event/${jobId}/cancel`,
+ endpoint: `/api/v1/task/${taskId}/cancel`,
}).catch(error => {
- logging.warn('Failed to cancel async job', jobId, error);
+ logging.warn('Failed to cancel task', taskId, error);
});
};
-export const waitForAsyncData = async (
- asyncResponse: AsyncEvent,
- signal?: AbortSignal,
-) =>
- new Promise((resolve, reject) => {
- const jobId = asyncResponse.job_id;
-
- let onAbort: (() => void) | undefined;
- const cleanup = () => {
- removeListener(jobId);
- if (onAbort && signal) {
- signal.removeEventListener('abort', onAbort);
- }
- };
-
- // Bail immediately if the caller has already aborted (e.g. the chart was
- // unmounted before the job started), avoiding a leaked listener.
- if (signal?.aborted) {
- cancelAsyncJob(jobId);
- reject(new DOMException('Aborted', 'AbortError'));
- return;
- }
-
- const listener = async (asyncEvent: AsyncEvent) => {
- switch (asyncEvent.status) {
- case JOB_STATUS.DONE: {
- // Forward the signal so the cached-result download is cancelled too
if
- // the caller aborts mid-fetch, rather than wasting
network/processing.
- let { data, status } = await fetchCachedData(asyncEvent, signal); //
eslint-disable-line prefer-const
- data = ensureIsArray(data);
- if (status === 'success') {
- resolve(data);
- } else {
- reject(data);
- }
- // Terminal status: the promise is settled, so fully clean up.
- cleanup();
- break;
- }
- case JOB_STATUS.ERROR: {
- const err = parseErrorJson(asyncEvent);
- reject(err);
- // Terminal status: the promise is settled, so fully clean up.
- cleanup();
- break;
- }
- default: {
- // Non-terminal status (e.g., 'pending', 'running'): keep the
listener
- // registered so it can receive the eventual terminal event ('done',
'error').
- // Only cleanup happens on terminal states or abort.
- logging.info(
- 'received non-terminal event with status',
- asyncEvent.status,
- );
- }
- }
- };
-
- // When the caller aborts (Stop pressed, chart superseded/unmounted), stop
- // listening so the listener and its retained closure don't leak, and ask
the
- // server to cancel the job so it stops consuming warehouse resources.
- if (signal) {
- onAbort = () => {
- cleanup();
- cancelAsyncJob(jobId);
- reject(new DOMException('Aborted', 'AbortError'));
- };
- signal.addEventListener('abort', onAbort, { once: true });
- }
-
- addListener(jobId, listener);
- });
-
-const fetchEvents = makeApi<
- { last_id?: string | null },
- { result: AsyncEvent[] }
->({
- method: 'GET',
- endpoint: POLLING_URL,
-});
-
-const setLastId = (asyncEvent: AsyncEvent) => {
- lastReceivedEventId = asyncEvent.id;
- try {
- localStorage.setItem(LOCALSTORAGE_KEY, lastReceivedEventId as string);
- } catch (err) {
- logging.warn('Error saving event Id to localStorage', err);
+const settle = (waiter: Waiter) => {
+ if (waiter.signal && waiter.onAbort) {
+ waiter.signal.removeEventListener('abort', waiter.onAbort);
+ }
+ if (waiter.failed) {
+ waiter.reject(
+ new Error('One or more chart-data queries failed'), // surfaced via
getClientErrorObject
+ );
+ } else {
+ waiter.resolve();
}
};
-export const processEvents = async (events: AsyncEvent[]) => {
- events.forEach((asyncEvent: AsyncEvent) => {
- const jobId = asyncEvent.job_id;
- const listener = listenersByJobId.get(jobId);
- // `jobId` originates from server/WebSocket payloads, so the listener is
- // resolved exclusively through a Map (never plain-object property access,
- // which would expose the prototype chain), and we confirm the retrieved
- // value is a registered function before dispatching the event to it.
- if (typeof listener === 'function') {
- listener(asyncEvent);
- retriesByJobId.delete(jobId);
- } else {
- // handle race condition where event is received
- // before listener is registered
- const retries = (retriesByJobId.get(jobId) ?? 0) + 1;
- retriesByJobId.set(jobId, retries);
-
- if (retries <= MAX_RETRIES) {
- setTimeout(() => {
- processEvents([asyncEvent]);
- }, RETRY_DELAY * retries);
- } else {
- retriesByJobId.delete(jobId);
- logging.warn('listener not found for job_id', asyncEvent.job_id);
- }
- }
- setLastId(asyncEvent);
- });
-};
-
-const getPollingDelay = () => {
- if (!consecutivePollingErrorCount) return pollingDelayMs;
- const backoffDelayMs = pollingDelayMs * 2 ** consecutivePollingErrorCount;
- return Math.max(
- pollingDelayMs,
- Math.min(backoffDelayMs, MAX_ERROR_POLLING_DELAY_MS),
- );
+const applyStatus = (taskId: string, status: string) => {
+ const waiter = waitersByTaskId.get(taskId);
+ if (!waiter || !TERMINAL_STATUSES.has(status)) return;
+ waitersByTaskId.delete(taskId);
+ waiter.pending.delete(taskId);
+ if (status !== STATUS_SUCCESS) waiter.failed = true;
+ if (waiter.pending.size === 0) settle(waiter);
};
-const loadEventsFromApi = async () => {
- const generation = pollingGeneration;
- const eventArgs = lastReceivedEventId ? { last_id: lastReceivedEventId } :
{};
- if (listenersByJobId.size) {
+const loadStatusChanges = async (generation: number) => {
+ if (generation !== pollingGeneration) return;
+ if (waitersByTaskId.size) {
try {
- const { result: events } = await fetchEvents(eventArgs);
+ const { statuses, cursor: next } = await fetchStatusChanges({
+ cursor,
+ task_type: CHART_QUERY_TASK_TYPE,
+ });
if (generation !== pollingGeneration) return;
- consecutivePollingErrorCount = 0;
- if (events?.length) await processEvents(events);
+ cursor = next;
+ Object.entries(statuses).forEach(([taskId, { status }]) =>
+ applyStatus(taskId, status),
+ );
} catch (err) {
if (generation !== pollingGeneration) return;
- consecutivePollingErrorCount += 1;
logging.warn(err);
}
}
-
- if (generation !== pollingGeneration) return;
- if (transport === TRANSPORT_POLLING) {
- pollingTimeoutId = window.setTimeout(loadEventsFromApi, getPollingDelay());
- }
+ // Reschedule from the tail so a slow request never overlaps the next tick.
+ pollingTimeoutId = window.setTimeout(
+ () => loadStatusChanges(generation),
+ pollingDelayMs,
+ );
};
-const wsConnectMaxRetries = 6;
-const wsConnectErrorDelay = 2500;
-let wsConnectRetries = 0;
-let wsConnectTimeout: any;
-let ws: WebSocket;
-
-const wsConnect = (): void => {
- let url = config.GLOBAL_ASYNC_QUERIES_WEBSOCKET_URL;
- if (lastReceivedEventId) url += `?last_id=${lastReceivedEventId}`;
- ws = new WebSocket(url);
-
- ws.addEventListener('open', () => {
- logging.log('WebSocket connected');
- clearTimeout(wsConnectTimeout);
- wsConnectRetries = 0;
- });
-
- ws.addEventListener('close', () => {
- wsConnectTimeout = setTimeout(() => {
- wsConnectRetries += 1;
- if (wsConnectRetries <= wsConnectMaxRetries) {
- wsConnect();
- } else {
- logging.warn('WebSocket not available, falling back to async polling');
- loadEventsFromApi();
- }
- }, wsConnectErrorDelay);
- });
-
- ws.addEventListener('error', () => {
- // https://developer.mozilla.org/en-US/docs/Web/API/WebSocket/readyState
- if (ws.readyState < 2) ws.close();
- });
+/**
+ * Await completion of an async chart-data job's query tasks, then re-issue the
+ * original request to read the now-cached results.
+ *
+ * Resolves with the fresh `QueryData[]` once every task has succeeded (the
+ * caller's `refetch` returns synchronously from the warm per-query cache);
+ * rejects if any task ends in a non-success terminal state, or with an
+ * AbortError if the caller aborts (which also cancels the outstanding tasks).
+ */
+export const waitForAsyncData = async <T = unknown[]>(
+ asyncJob: AsyncJob,
+ refetch: () => Promise<T>,
+ signal?: AbortSignal,
+): Promise<T> => {
+ const taskIds = asyncJob.task_ids ?? [];
+ if (baselineReady) await baselineReady;
- ws.addEventListener('message', async event => {
- let events: AsyncEvent[] = [];
- try {
- events = [JSON.parse(event.data)];
- await processEvents(events);
- } catch (err) {
- logging.warn(err);
+ await new Promise<void>((resolve, reject) => {
+ if (signal?.aborted) {
+ taskIds.forEach(cancelTask);
+ reject(new DOMException('Aborted', 'AbortError'));
+ return;
}
+ const waiter: Waiter = {
+ pending: new Set(taskIds),
+ failed: false,
+ resolve,
+ reject,
+ signal,
+ };
+ if (signal) {
+ waiter.onAbort = () => {
+ waiter.pending.forEach(taskId => waitersByTaskId.delete(taskId));
+ taskIds.forEach(cancelTask);
+ reject(new DOMException('Aborted', 'AbortError'));
+ };
+ signal.addEventListener('abort', waiter.onAbort, { once: true });
+ }
+ if (!taskIds.length) {
+ settle(waiter);
+ return;
+ }
+ taskIds.forEach(taskId => waitersByTaskId.set(taskId, waiter));
Review Comment:
**Suggestion:** The task registry stores only one `Waiter` per task ID.
Because shared GTF tasks can be returned to multiple concurrent requests,
registering a later waiter overwrites the earlier one; when the task completes,
only the latest request is settled and the earlier request remains pending
indefinitely. Store a set/list of waiters per task ID and settle all of them.
[race condition]
<details>
<summary><b>Severity Level:</b> Critical 🚨</summary>
```mdx
- ❌ Concurrent charts sharing tasks can remain permanently loading.
- ❌ Earlier requests never refetch completed chart data.
- ⚠️ Shared-task deduplication becomes user-visible request starvation.
```
</details>
[](https://docs.codeant.ai/cli/resolve-pr-comments-skill)
<details>
<summary><b>Prompt for AI Agent 🤖 </b></summary>
```mdx
This is a comment left during a code review.
**Path:** superset-frontend/src/middleware/asyncEvent.ts
**Line:** 188:188
**Comment:**
*Race Condition: The task registry stores only one `Waiter` per task
ID. Because shared GTF tasks can be returned to multiple concurrent requests,
registering a later waiter overwrites the earlier one; when the task completes,
only the latest request is settled and the earlier request remains pending
indefinitely. Store a set/list of waiters per task ID and settle all of them.
Validate the correctness of the flagged issue. If correct, How can I resolve
this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask
user if the user wants to fix the rest of the comments as well. if said yes,
then fetch all the comments validate the correctness and implement a minimal fix
```
</details>
<a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F43424&comment_hash=f813a4ce5b7f500b18eac1a773e4d55d263d913983028b16c0c4f8805f1c7ec4&reaction=like'>👍</a>
| <a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F43424&comment_hash=f813a4ce5b7f500b18eac1a773e4d55d263d913983028b16c0c4f8805f1c7ec4&reaction=dislike'>👎</a>
##########
superset/tasks/async_queries.py:
##########
@@ -16,126 +16,185 @@
# under the License.
from __future__ import annotations
-import dataclasses
import logging
from typing import Any, TYPE_CHECKING
-from celery.exceptions import SoftTimeLimitExceeded
from flask import current_app
from flask_appbuilder.security.sqla.models import User
-from marshmallow import ValidationError
+from superset_core.tasks.types import TaskOptions, TaskScope
-from superset.charts.data.form_data import set_form_data
-from superset.charts.schemas import ChartDataQueryContextSchema
-from superset.exceptions import (
- SupersetErrorException,
- SupersetErrorsException,
+from superset.common.query_serialization import (
+ load_serialized_query,
+ serialize_query,
+ SerializedQuery,
)
+from superset.constants import CacheRegion
from superset.extensions import (
- async_query_manager,
- celery_app,
security_manager,
)
+from superset.tasks.decorators import task
from superset.utils.core import override_user
-from superset.utils.error_sanitization import sanitize_error_dicts
if TYPE_CHECKING:
from superset.common.query_context import QueryContext
+ from superset.common.query_object import QueryObject
+ from superset.models.tasks import Task
+ from superset.security.guest_token import GuestToken
logger = logging.getLogger(__name__)
query_timeout = current_app.config[
"SQLLAB_ASYNC_TIME_LIMIT_SEC"
] # TODO: new config key
+# GTF task type for the chart-data fan-out. Each QueryObject runs as its own
SHARED
+# task keyed by its query_cache_key (safe cross-user dedup — the key encodes
+# RLS/impersonation). The client polls /api/v1/task/status_changes (filtered
to this
+# type) and aggregates the tasks' statuses itself; GTF owns completion
emission (there
+# is no coordinator task). The atomic unit is a QueryObject, not
chart-specific, so the
+# type is versioned to allow the serialization/execution contract to evolve.
+CHART_QUERY_TASK = "superset.query_object_v1"
-def _create_query_context_from_form(form_data: dict[str, Any]) -> QueryContext:
- """
- Create the query context from the form data.
-
- :param form_data: The task form data
- :returns: The query context
- :raises ValidationError: If the request is incorrect
- """
-
- try:
- return ChartDataQueryContextSchema().load(form_data)
- except KeyError as ex:
- raise ValidationError("Request is incorrect") from ex
+def _resolve_user(user_id: int | None, guest_token: "GuestToken | None") ->
User:
+ """Resolve the acting user for an async chart-data task.
-def _load_user_from_job_metadata(job_metadata: dict[str, Any]) -> User:
- if user_id := job_metadata.get("user_id"):
- # logged in user
- user = security_manager.get_user_by_id(user_id)
- elif guest_token := job_metadata.get("guest_token"):
- # embedded guest user
- user = security_manager.get_guest_user_from_token(guest_token)
- del job_metadata["guest_token"]
- else:
- # default to anonymous user if no user is found
- user = security_manager.get_anonymous_user()
- return user
+ The GTF executor does not impersonate on its own, so each task establishes
the
+ request user itself (for RLS/impersonation), mirroring the legacy Celery
path.
+ """
+ if user_id:
+ return security_manager.get_user_by_id(user_id)
+ if guest_token:
+ return security_manager.get_guest_user_from_token(guest_token)
+ return security_manager.get_anonymous_user()
-def _handle_soft_time_limit(
- job_metadata: dict[str, Any], ex: Exception, activity: str
+def _inject_contribution_totals(
+ query_obj: "QueryObject", totals_cache_key: str
) -> None:
+ """Inject ``contribution_totals`` from the cached totals query into
``query_obj``.
+
+ A contribution query normalizes its metrics against column sums from a
separate
+ "totals" query. In the per-query task model the totals query runs as its
own
+ task (a ``depends_on`` prerequisite) and caches its dataframe; here we
read that
+ cached dataframe and inject the sums into this query's contribution
+ post-processing before it runs — the same result the synchronous
+ ``ensure_totals_available`` produces, but reading the cache the
prerequisite
+ populated instead of re-running the totals query. ``contribution_totals``
is
+ stripped from the cache key, so this affects only the result, not the key.
"""
- SoftTimeLimitExceeded is raised both by a genuine timeout and by a
- user-initiated cancel (revoke sends SIGUSR1). The cancel endpoint has
- already emitted the terminal event for the latter - it has to, since a task
- revoked while still queued never reaches this handler - so only a timeout
- is reported here, and without one the client would wait forever.
- """
- if async_query_manager.is_job_cancelled(job_metadata["job_id"]):
- logger.info("Cancelled by the user while %s", activity)
+ from superset.common.utils.query_cache_manager import QueryCacheManager
+
+ cache = QueryCacheManager.get(key=totals_cache_key,
region=CacheRegion.DATA)
+ if not cache.is_loaded or cache.df is None:
+ # The depends_on prerequisite guarantees the totals task succeeded, so
a
+ # miss here is unexpected; leave the query as-is (the contribution op
will
+ # fall back to its own totals) rather than failing the whole chart.
+ logger.warning(
+ "Totals result not cached under %s; contribution left
un-normalized",
+ totals_cache_key,
+ )
return
Review Comment:
**Suggestion:** When the totals cache is unavailable, this branch marks the
contribution task as successful and executes it without injecting
`contribution_totals`. The reconstructed context contains only the contribution
query, so it cannot perform the synchronous path's `ensure_totals_available`
operation; the result is therefore unnormalized or fails later while the task
is still reported as successful. Treat a missing prerequisite cache as task
failure instead of silently returning. [incomplete implementation]
<details>
<summary><b>Severity Level:</b> Major ⚠️</summary>
```mdx
- ❌ Contribution charts can display incorrectly normalized percentages.
- ⚠️ Successful task status hides missing prerequisite data.
- ⚠️ Frontend re-request returns incorrect cached chart results.
```
</details>
[](https://docs.codeant.ai/cli/resolve-pr-comments-skill)
<details>
<summary><b>Prompt for AI Agent 🤖 </b></summary>
```mdx
This is a comment left during a code review.
**Path:** superset/tasks/async_queries.py
**Line:** 88:96
**Comment:**
*Incomplete Implementation: When the totals cache is unavailable, this
branch marks the contribution task as successful and executes it without
injecting `contribution_totals`. The reconstructed context contains only the
contribution query, so it cannot perform the synchronous path's
`ensure_totals_available` operation; the result is therefore unnormalized or
fails later while the task is still reported as successful. Treat a missing
prerequisite cache as task failure instead of silently returning.
Validate the correctness of the flagged issue. If correct, How can I resolve
this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask
user if the user wants to fix the rest of the comments as well. if said yes,
then fetch all the comments validate the correctness and implement a minimal fix
```
</details>
<a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F43424&comment_hash=18538afe5361e9bd624ef88a016f4fa9e139d2e09289b20bae85a70f25ddb4ee&reaction=like'>👍</a>
| <a
href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F43424&comment_hash=18538afe5361e9bd624ef88a016f4fa9e139d2e09289b20bae85a70f25ddb4ee&reaction=dislike'>👎</a>
--
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]
