You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@superset.apache.org by el...@apache.org on 2026/10/22 00:31:48 UTC
(superset) 04/04: fix(async): leave contribution queries to the sync follow-up when totals are not cached
This is an automated email from the ASF dual-hosted git repository.
eschutho pushed a commit to branch fix-cache-max-value-size-default
in repository https://gitbox.apache.org/repos/asf/superset.git
commit e00514eecf06e74fb3365a93ee84953162b8ab35
Author: Elizabeth Thompson <es...@gmail.com>
AuthorDate: Wed Oct 7 20:59:33 2026 +0000
fix(async): leave contribution queries to the sync follow-up when totals are not cached
An async contribution chart runs its totals query as a prerequisite task
and reads that task's result from the data cache. When the totals result
is larger than DATA_CACHE_MAX_VALUE_SIZE it is not cached, so the
contribution task raised "Contribution totals not found in cache" and the
chart failed instead of reaching the client's synchronous follow-up
request.
When the totals are not in the cache, the contribution task now finishes
without running or caching its query and without publishing a cache key.
The synchronous follow-up computes the totals itself
(ensure_totals_available) and runs the contribution query, so the chart
renders. No contribution result computed without its totals is cached.
Add a test that runs the real task body for the totals task and the
dependent contribution task against a real cache, for totals that fit and
totals that are too large.
Co-Authored-By: Claude Opus 5.5 <no...@anthropic.com>
---
UPDATING.md | 4 +-
docs/admin_docs/configuration/cache.mdx | 2 +
superset/tasks/async_queries.py | 32 ++++--
.../common/test_oversized_result_readback.py | 114 ++++++++++++++++++++-
4 files changed, 136 insertions(+), 16 deletions(-)
diff --git a/UPDATING.md b/UPDATING.md
index dd7341b9399..2fc3ddc7e37 100644
--- a/UPDATING.md
+++ b/UPDATING.md
@@ -671,7 +671,9 @@ the next time it is requested. This includes the follow-up request a chart sends
after a background (async) query finishes: that request recomputes the result
instead of reading it from the cache, so an oversized async chart runs its query
twice. That recompute happens inside the web request itself, so a very slow, very
-large query is subject to the web server's request timeout.
+large query is subject to the web server's request timeout. For an async chart
+that shows contributions (percent of total), if the totals result is too large to
+cache, the follow-up request computes the totals and the chart in the same way.
The limit applies to every writer to the data cache: chart results, SQL query
results from the SQL executor, filter dropdown column values, and compatible
diff --git a/docs/admin_docs/configuration/cache.mdx b/docs/admin_docs/configuration/cache.mdx
index a7b84823306..56104fe4a9d 100644
--- a/docs/admin_docs/configuration/cache.mdx
+++ b/docs/admin_docs/configuration/cache.mdx
@@ -198,6 +198,8 @@ returns its data, and the value is recomputed from the datasource the next time
With `GLOBAL_ASYNC_QUERIES`, that includes the follow-up request a chart sends after its background
query finishes, so an oversized async chart runs its query twice. That recompute happens inside the
web request itself, so a very slow, very large query is subject to the web server's request timeout.
+For an async chart that shows contributions (percent of total), if the totals result is too large to
+cache, the follow-up request computes the totals and the chart in the same way.
Each skip emits a
WARNING log naming the key and byte size, and increments the `skip_cache_value_too_large` statsd
metric. Set to `None` to disable the check entirely (no serialization overhead is then incurred).
diff --git a/superset/tasks/async_queries.py b/superset/tasks/async_queries.py
index 7b2dcc5f828..aacddc67ce7 100644
--- a/superset/tasks/async_queries.py
+++ b/superset/tasks/async_queries.py
@@ -177,7 +177,7 @@ def _resolve_user(user_id: int | None, guest_token: "GuestToken | None") -> User
def _inject_contribution_totals(
query_obj: "QueryObject", totals_cache_key: str
-) -> None:
+) -> bool:
"""Inject ``contribution_totals`` from the cached totals query into ``query_obj``.
A contribution query normalizes its metrics against column sums from a separate
@@ -188,26 +188,24 @@ def _inject_contribution_totals(
``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.
+
+ :returns: ``False`` when the totals are not in the cache, in which case nothing
+ is injected. The totals task can succeed without caching its result: a value
+ larger than ``DATA_CACHE_MAX_VALUE_SIZE`` is not stored, and an entry can be
+ evicted before this task reads it.
"""
from superset.common.query_context_processor import is_summable
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 and wrote
- # this cache entry, so a miss is unexpected (e.g. it was evicted between the
- # totals task finishing and this task reading). Fail loudly rather than
- # caching a silently un-normalized result the client would then re-request:
- # this task's single query cannot reproduce the synchronous path's
- # ensure_totals_available (it has no totals query to run).
- raise SupersetException(
- f"Contribution totals not found in cache under {totals_cache_key}"
- )
+ return False
df = cache.df
totals = {col: df[col].sum() for col in df.columns if is_summable(df[col])}
for post_processing in query_obj.post_processing or []:
if post_processing.get("operation") == "contribution":
post_processing.setdefault("options", {})["contribution_totals"] = totals
+ return True
def _get_dependency_cache_key() -> str:
@@ -324,7 +322,19 @@ def execute_chart_query(
# task reads back under the same id. Only consulted when force is true.
query_obj.force_nonce = str(get_context().task_uuid)
if requires_totals:
- _inject_contribution_totals(query_obj, _get_dependency_cache_key())
+ totals_cache_key = _get_dependency_cache_key()
+ if not _inject_contribution_totals(query_obj, totals_cache_key):
+ # Without the totals this task cannot produce a correct result,
+ # and this query alone cannot recompute them. Finish without
+ # running or caching the query and without publishing a cache key:
+ # the client's synchronous follow-up request then computes the
+ # totals itself (``ensure_totals_available``) and runs this query.
+ logger.warning(
+ "Contribution totals not found in cache under %s; "
+ "leaving the query to the synchronous follow-up request",
+ totals_cache_key,
+ )
+ return
# Executes on cache miss and writes CacheRegion.DATA under query_cache_key.
try:
with _capture_query_cancellation(query_context):
diff --git a/tests/unit_tests/common/test_oversized_result_readback.py b/tests/unit_tests/common/test_oversized_result_readback.py
index 45f806e6655..7fb7878c43a 100644
--- a/tests/unit_tests/common/test_oversized_result_readback.py
+++ b/tests/unit_tests/common/test_oversized_result_readback.py
@@ -22,10 +22,13 @@ re-sends the same request synchronously. When the result is larger than
nothing in the cache and runs the query again. These tests drive
``get_df_payload_result`` twice against a real in-memory cache -- once as the
background task, once as the follow-up -- for a normal load and for a forced
-refresh whose follow-up carries the task's id as its ``force_nonce``.
+refresh whose follow-up carries the task's id as its ``force_nonce``. The
+contribution tests run the real async task body for a totals task and the
+contribution task that depends on it.
"""
from collections.abc import Iterator
+from contextlib import nullcontext
from typing import Any
from unittest.mock import MagicMock
@@ -65,8 +68,11 @@ def data_cache(mocker: MockerFixture) -> Iterator[Cache]:
cache.clear()
-def _processor(mocker: MockerFixture, rows: int, force: bool) -> Any:
- """A processor whose datasource returns a ``rows``-row dataframe."""
+def _processor(
+ mocker: MockerFixture, rows: int, force: bool, cache_key: str = CACHE_KEY
+) -> Any:
+ """A processor whose datasource returns a ``rows``-row dataframe, cached
+ under ``cache_key``."""
query_context = MagicMock()
query_context.force = force
query_context.force_nonce = None
@@ -85,7 +91,7 @@ def _processor(mocker: MockerFixture, rows: int, force: bool) -> Any:
query_result.semantic_cache_status = None
mocker.patch.object(processor, "get_query_result", return_value=query_result)
mocker.patch.object(processor, "get_annotation_data", return_value={})
- mocker.patch.object(processor, "query_cache_key", return_value=CACHE_KEY)
+ mocker.patch.object(processor, "query_cache_key", return_value=cache_key)
mocker.patch.object(processor, "get_cache_timeout", return_value=300)
return processor
@@ -171,3 +177,103 @@ def test_oversized_refresh_removes_older_cached_result(
processor.get_query_result.assert_called_once()
assert not payload["is_cached"]
assert payload["rowcount"] == 2000
+
+
+def _run_chart_task(
+ mocker: MockerFixture,
+ processor: Any,
+ query_obj: Any,
+ requires_totals: bool = False,
+ dependency_payloads: list[dict[str, Any]] | None = None,
+) -> MagicMock:
+ """Run the real ``execute_chart_query`` task body for one query.
+
+ Only the worker plumbing (user, task context, form data, cancellation hook) is
+ stubbed; the query runs through ``processor`` against the real cache.
+
+ :returns: the task context, whose ``update_task`` records the published payload
+ """
+ from superset.tasks.async_queries import execute_chart_query
+
+ query_context = MagicMock()
+ query_context.queries = [query_obj]
+ query_context.get_df_payload_result.side_effect = processor.get_df_payload_result
+ task_context = MagicMock(task_uuid=TASK_ID)
+ task_context.get_dependency_payloads.return_value = dependency_payloads or []
+ mocker.patch("superset.tasks.async_queries._resolve_user", return_value=MagicMock())
+ mocker.patch("superset.tasks.async_queries.override_user")
+ mocker.patch(
+ "superset.tasks.async_queries.load_serialized_query",
+ return_value=query_context,
+ )
+ mocker.patch("superset.tasks.async_queries.get_context", return_value=task_context)
+ mocker.patch("superset.charts.data.form_data.set_query_context_form_data")
+ mocker.patch(
+ "superset.tasks.async_queries._capture_query_cancellation",
+ return_value=nullcontext(),
+ )
+ execute_chart_query.func(
+ {"datasource": {}, "query": {}}, # type: ignore[typeddict-item]
+ user_id=7,
+ requires_totals=requires_totals,
+ )
+ return task_context
+
+
+def _contribution_query_object() -> Any:
+ from superset.common.query_object import QueryObject
+
+ query_obj = QueryObject(
+ datasource=MagicMock(),
+ columns=["name"],
+ post_processing=[{"operation": "contribution", "options": {}}],
+ )
+ query_obj.force_nonce = None
+ return query_obj
+
+
+@pytest.mark.parametrize(
+ "totals_rows,totals_cached", [(2, True), (2000, False)], ids=["fits", "oversized"]
+)
+def test_contribution_task_after_totals_task(
+ mocker: MockerFixture, data_cache: Cache, totals_rows: int, totals_cached: bool
+) -> None:
+ """The async totals task feeds the contribution task through the data cache.
+
+ When the totals result fits, the contribution task reads it, runs, and
+ publishes its own cache key. When the totals result is too large to cache, the
+ contribution task must not fail: it leaves its query uncached and publishes
+ nothing, so the client's synchronous follow-up request (which computes the
+ totals itself in ``ensure_totals_available``) renders the chart.
+ """
+ totals = _processor(mocker, totals_rows, force=False, cache_key="totals-key")
+ totals_task = _run_chart_task(mocker, totals, _query_object(None))
+ published = totals_task.update_task.call_args.kwargs["payload"]
+ assert published == {"cache_key": "totals-key"}
+ assert (data_cache.get("totals-key") is not None) is totals_cached
+
+ contribution = _processor(mocker, 2, force=False, cache_key="contribution-key")
+ contribution_task = _run_chart_task(
+ mocker,
+ contribution,
+ _contribution_query_object(),
+ requires_totals=True,
+ dependency_payloads=[published],
+ )
+
+ if totals_cached:
+ contribution.get_query_result.assert_called_once()
+ contribution_task.update_task.assert_called_once_with(
+ payload={"cache_key": "contribution-key"}, immediate=True
+ )
+ assert data_cache.get("contribution-key") is not None
+ return
+
+ contribution.get_query_result.assert_not_called()
+ contribution_task.update_task.assert_not_called()
+ assert data_cache.get("contribution-key") is None
+ # The synchronous follow-up misses the cache and computes the query.
+ payload = contribution.get_df_payload_result(_contribution_query_object()).payload
+ contribution.get_query_result.assert_called_once()
+ assert payload["error"] is None
+ assert payload["rowcount"] == 2