rebenitez1802 commented on code in PR #45561:
URL: https://github.com/apache/superset/pull/45561#discussion_r4341671580
##########
superset/sql/execution/executor.py:
##########
@@ -680,16 +684,24 @@ def execute(
QueryStatus.TIMED_OUT,
"Query exceeded the timeout limit",
start_time,
+ query=query,
)
except SupersetSecurityException as ex:
- return self._create_error_result(QueryStatus.FAILED, str(ex),
start_time)
- except (OAuth2RedirectError, OAuth2Error):
+ return self._create_error_result(
+ QueryStatus.FAILED, str(ex), start_time, query=query
+ )
+ except (OAuth2RedirectError, OAuth2Error) as ex:
# Let OAuth2 exceptions propagate so callers (MCP, API) can
# handle them with context-appropriate responses.
+ self._create_error_result(
+ QueryStatus.FAILED, str(ex), start_time, query=query
+ )
raise
except Exception as ex:
error_msg = self.database.db_engine_spec.extract_error_message(ex)
- return self._create_error_result(QueryStatus.FAILED, error_msg,
start_time)
+ return self._create_error_result(
+ QueryStatus.FAILED, error_msg, start_time, query=query
+ )
Review Comment:
This handler also catches failures from `_store_in_cache()` (line 678),
which runs *after* the success commit while `query` is still set — so a cache
error (e.g. Redis down) rewrites an already-committed `success` row to
`failed`. Suggest guarding the cache write like `_get_from_cache` does:
```python
if not has_mutation:
try:
self._store_in_cache(result, final_sql, opts)
except Exception: # pylint: disable=broad-except
logger.warning("Failed to cache query result", exc_info=True)
```
(or set `query = None` right after the success commit), plus a test.
##########
superset/sql/execution/executor.py:
##########
@@ -1008,14 +1021,41 @@ def _create_error_result(
:param start_time: Start time for calculating execution duration
:param partial_results: Optional list of StatementResult from
successful
statements before the failure
+ :param query: Persisted audit record, if execution reached that stage
:returns: QueryResult with error status
"""
- from superset_core.queries.types import QueryResult as QueryResultType
+ from superset_core.queries.types import (
+ QueryResult as QueryResultType,
+ QueryStatus,
+ )
+
+ from superset.sql.execution.utils import commit_query_unless_stopped
+
+ query_id = None
+ if query is not None:
+ try:
+ # Read the persisted identity without loading expired
attributes.
+ query_id = inspect(query).identity[0]
+ # A failed metadata flush must be rolled back before recording
failure.
+ if not db.session.is_active:
+ db.session.rollback() # pylint:
disable=consider-using-transaction
+ query.status = status.value
+ query.error_message = error_message
+ query.end_time = now_as_float()
+ if not commit_query_unless_stopped(query):
+ status = QueryStatus.STOPPED
Review Comment:
On a False return, `commit_query_unless_stopped()` has already run
`refresh(query, with_for_update=True)` without committing, so the Query row
lock stays held until session teardown. All other call sites commit right after
(`celery_task.py:609`, `sql_lab.py:160/503/759/840`):
```suggestion
if not commit_query_unless_stopped(query):
status = QueryStatus.STOPPED
error_message = query.error_message
# Release the row lock taken by the locking refresh.
db.session.commit() # pylint:
disable=consider-using-transaction
```
##########
superset/sql/execution/executor.py:
##########
@@ -1008,14 +1021,41 @@ def _create_error_result(
:param start_time: Start time for calculating execution duration
:param partial_results: Optional list of StatementResult from
successful
statements before the failure
+ :param query: Persisted audit record, if execution reached that stage
:returns: QueryResult with error status
"""
- from superset_core.queries.types import QueryResult as QueryResultType
+ from superset_core.queries.types import (
+ QueryResult as QueryResultType,
+ QueryStatus,
+ )
+
+ from superset.sql.execution.utils import commit_query_unless_stopped
+
+ query_id = None
+ if query is not None:
+ try:
+ # Read the persisted identity without loading expired
attributes.
+ query_id = inspect(query).identity[0]
+ # A failed metadata flush must be rolled back before recording
failure.
+ if not db.session.is_active:
+ db.session.rollback() # pylint:
disable=consider-using-transaction
+ query.status = status.value
+ query.error_message = error_message
Review Comment:
nit: celery/legacy use `query.end_time or now_as_float()` — matching that
keeps the paths consistent.
```suggestion
query.end_time = query.end_time or now_as_float()
```
--
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]
