You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@superset.apache.org by hu...@apache.org on 2026/10/21 08:04:28 UTC
(superset) branch hughhhh/pfm-probe-bounds created (now 00d16072c41)
This is an automated email from the ASF dual-hosted git repository.
hughhhh pushed a change to branch hughhhh/pfm-probe-bounds
in repository https://gitbox.apache.org/repos/asf/superset.git
at 00d16072c41 fix(datasets): bound the partition transform probe
This branch includes the following new commits:
new 00d16072c41 fix(datasets): bound the partition transform probe
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(superset) 01/01: fix(datasets): bound the partition transform probe
Posted by hu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hughhhh pushed a commit to branch hughhhh/pfm-probe-bounds
in repository https://gitbox.apache.org/repos/asf/superset.git
commit 00d16072c41fe2187be3229bd4d846d1e6620c07
Author: Hugh A Miles II <hu...@gmail.com>
AuthorDate: Wed Oct 21 09:04:15 2026 +0100
fix(datasets): bound the partition transform probe
The synchronous probe on the chart-query path had no upper bound on its
work: a failing transform was re-probed on every query, a wide IN list
became one SELECT column per value (PostgreSQL refuses a 1,665th), nothing
limited its runtime, and under the default NullCache even successful
probes re-ran on every chart load.
- PARTITION_TRANSFORM_PROBE_MAX_VALUES (1000): decline wider filters.
- PARTITION_TRANSFORM_PROBE_TIMEOUT (10s): utils.timeout around get_df.
- PARTITION_TRANSFORM_PROBE_FAILURE_CACHE_TIMEOUT (5m): remember failures,
with the engine's messages so the preview can still explain them.
- Fall back to DATA_CACHE_CONFIG, then a bounded per-process cache, when
CACHE_CONFIG is a NullCache.
- partition_mapping.probe.* outcome counters and latency_ms timing.
Co-Authored-By: Claude Opus 5.5 <no...@anthropic.com>
---
.../configuration/partition-filter-mapping.mdx | 46 +++--
superset/config.py | 18 ++
superset/connectors/sqla/partition_mapping.py | 207 ++++++++++++++++++---
tests/unit_tests/conftest.py | 14 ++
.../connectors/sqla/partition_mapping_test.py | 166 ++++++++++++++++-
5 files changed, 409 insertions(+), 42 deletions(-)
diff --git a/docs/admin_docs/configuration/partition-filter-mapping.mdx b/docs/admin_docs/configuration/partition-filter-mapping.mdx
index 1190a7e4d43..8c385421f0e 100644
--- a/docs/admin_docs/configuration/partition-filter-mapping.mdx
+++ b/docs/admin_docs/configuration/partition-filter-mapping.mdx
@@ -272,7 +272,22 @@ adds a round trip to the chart query path. Day-aligned ranges like "Last month"
constantly; second-granularity relative ranges like "Last 24 hours" essentially never do.
If the evaluation fails for any reason, no predicate is added: the query still runs and is
-still correct, it just scans more partitions.
+still correct, it just scans more partitions. A failure is remembered for
+`PARTITION_TRANSFORM_PROBE_FAILURE_CACHE_TIMEOUT` (5 minutes by default), so a transform the
+database always rejects is not re-run on every chart load, and a transient error costs
+minutes of pruning rather than a day.
+
+The evaluation is bounded. Every distinct filter value is one column of a single `SELECT`,
+and engines cap the select list — PostgreSQL refuses more than 1,664 entries — so a filter
+with more than `PARTITION_TRANSFORM_PROBE_MAX_VALUES` distinct values (1,000 by default) is
+not evaluated and its query does not prune. An evaluation that runs longer than
+`PARTITION_TRANSFORM_PROBE_TIMEOUT` (10 seconds) is abandoned the same way. That timeout is
+best effort: like Superset's other query timeouts it is enforced with `SIGALRM`, so only in
+a worker that serves requests on its main thread.
+
+Each evaluation reports to the configured `STATS_LOGGER` under `partition_mapping.probe.`:
+`cache_hit`, `cache_miss`, `failure_cache_hit`, `success`, `error`, `timeout`,
+`declined.<reason>`, and a `latency_ms` timing.
The same goes for a transform that evaluates but answers with something the partition column
cannot hold — a number against a text key, or text against a numeric one. Not every engine
@@ -324,26 +339,25 @@ column resolved back to it.
| Setting | Default | Purpose |
|---|---|---|
| `PARTITION_TRANSFORM_PROBE_CACHE_TIMEOUT` | 24 hours | How long an evaluated transform stays cached |
+| `PARTITION_TRANSFORM_PROBE_FAILURE_CACHE_TIMEOUT` | 5 minutes | How long a failed evaluation is remembered before the database is asked again; 0 retries on every query |
+| `PARTITION_TRANSFORM_PROBE_MAX_VALUES` | 1,000 | Most distinct filter values one evaluation covers; a wider filter does not prune. 0 for no cap |
+| `PARTITION_TRANSFORM_PROBE_TIMEOUT` | 10 seconds | How long one evaluation may run before it is abandoned; 0 to disable |
| `PARTITION_TRANSFORM_PREVIEW_RATE_LIMIT` | 30 | Per-user, per-dataset preview requests per minute; the editor's preview panel runs a real query |
-### Three things need a working cache backend
+### Configure a shared cache backend
-None of them does anything on its own. The probe cache, the preview budget and
-the pruning indicator all live in **`CACHE_CONFIG`**, whose default is
-`{"CACHE_TYPE": "NullCache"}` — so out of the box:
+The probe cache, the preview budget and the pruning indicator all live in
+**`CACHE_CONFIG`**, whose default is `{"CACHE_TYPE": "NullCache"}`.
-- every chart load on a mapped dataset re-probes the warehouse, because nothing
- is cached;
-- the preview budget is never enforced, because a null cache reports every
- request as the first one in its window; and
-- the pruning indicator stays optimistic. A transform the database rejects
- parses fine, so Superset cannot tell it apart from a working one until
- something probes — and with nothing to record that answer in, the editor's
- banner and the glyph on a filter chip go on promising a speed-up the query
- gives up. Charts stay correct either way; what is lost is the UI's ability
- to stop claiming otherwise.
+The probe cache and the pruning indicator fall back to `DATA_CACHE_CONFIG` when
+`CACHE_CONFIG` is a null cache, and to a small per-process cache when both are.
+That keeps a chart load from re-probing the warehouse every time, but each
+gunicorn worker probes and records on its own, so the indicator can disagree
+between workers until each one has probed.
-Configuring `DATA_CACHE_CONFIG` does not help; all three read `CACHE_CONFIG`.
+The preview budget has no fallback: under a null `CACHE_CONFIG` it is never
+enforced, because a null cache reports every request as the first one in its
+window.
The backend also has to be **shared** across processes. A per-process
`SimpleCache` or a `FileSystemCache` on a local disk gives each gunicorn worker
diff --git a/superset/config.py b/superset/config.py
index a61d507bec5..d2c301af4ac 100644
--- a/superset/config.py
+++ b/superset/config.py
@@ -436,6 +436,24 @@ RATELIMIT_APPLICATION = "50 per second"
# second-granularity relative ranges ("Last 24 hours") essentially never.
PARTITION_TRANSFORM_PROBE_CACHE_TIMEOUT = int(timedelta(days=1).total_seconds())
+# How long a failed probe is remembered before the warehouse is asked again.
+# Short on purpose: long enough that a transform the engine always rejects is not
+# re-run on every chart load, short enough that a transient failure costs
+# minutes of pruning rather than a day. Set to 0 to retry on every query.
+PARTITION_TRANSFORM_PROBE_FAILURE_CACHE_TIMEOUT = int(
+ timedelta(minutes=5).total_seconds()
+)
+
+# Most distinct filter values one probe evaluates. Each value is one column of a
+# single SELECT, and engines cap the select list (PostgreSQL at 1,664); a wider
+# filter is not probed and its query does not prune. Set to 0 for no cap.
+PARTITION_TRANSFORM_PROBE_MAX_VALUES = 1000
+
+# Seconds a probe may run before it is abandoned and the query does not prune.
+# Best effort: enforced with SIGALRM, so only in a main-thread worker. Set to 0
+# to disable.
+PARTITION_TRANSFORM_PROBE_TIMEOUT = int(timedelta(seconds=10).total_seconds())
+
# Per-user, per-dataset budget for the partition mapping preview endpoint, which
# fires a real warehouse query from a text input in the dataset editor. Counted
# in a fixed 60-second window; set to 0 to disable.
diff --git a/superset/connectors/sqla/partition_mapping.py b/superset/connectors/sqla/partition_mapping.py
index bf2335877f4..11992254f39 100644
--- a/superset/connectors/sqla/partition_mapping.py
+++ b/superset/connectors/sqla/partition_mapping.py
@@ -182,7 +182,9 @@ import hashlib
import logging
import numbers
import re
+import time
from collections.abc import Sequence
+from contextlib import nullcontext
from dataclasses import dataclass
from datetime import date, datetime
from functools import lru_cache
@@ -194,6 +196,7 @@ import sqlalchemy as sa
from dateutil.relativedelta import relativedelta
from flask import current_app as app
from flask_babel import lazy_gettext as _
+from flask_caching.backends import NullCache, SimpleCache
from sqlalchemy.engine.interfaces import Dialect
from sqlalchemy.sql.elements import ColumnElement
@@ -202,8 +205,13 @@ from superset.exceptions import (
QueryClauseValidationException,
SupersetParseError,
SupersetSecurityException,
+ SupersetTimeoutException,
+)
+from superset.extensions import (
+ cache_manager,
+ feature_flag_manager,
+ stats_logger_manager,
)
-from superset.extensions import cache_manager, feature_flag_manager
from superset.sql.parse import SQLScript, SQLStatement
from superset.superset_typing import FilterValues
from superset.utils import core as utils, json
@@ -232,6 +240,17 @@ FEATURE_FLAG = "PARTITION_FILTER_MAPPING"
#: `update_from_object`.
MAX_TRANSFORM_LENGTH = 1024
+#: Prefix of every statsd key the probe emits: ``.cache_hit``, ``.cache_miss``,
+#: ``.failure_cache_hit``, ``.success``, ``.error``, ``.timeout``,
+#: ``.declined.<reason>`` and the ``.latency_ms`` timing.
+PROBE_METRIC_PREFIX = "partition_mapping.probe"
+
+#: Where probe results and verdicts go when neither configured cache keeps
+#: anything -- see `_probe_cache`. Per process, so each worker probes an entry
+#: once rather than once per chart: worse than a shared backend, far better
+#: than none. Bounded so a stream of distinct filter values cannot grow it.
+_LOCAL_PROBE_CACHE = SimpleCache(threshold=1024)
+
#: Placeholder the owner writes in the transform, e.g. ``unix_timestamp(:value)``.
#: Matched with word boundaries so ``:values`` is not mistaken for it.
VALUE_PLACEHOLDER_RE = re.compile(r":value\b")
@@ -1825,6 +1844,26 @@ def evaluate_transform(
seen.add(key)
distinct.append(value)
+ # Every value is one column of a single SELECT, and engines cap the select
+ # list: PostgreSQL refuses a 1,665th entry outright. Declining a filter that
+ # wide costs little pruning -- it spans that many partitions anyway -- and
+ # spares the warehouse a statement it would refuse on every chart load.
+ max_values = app.config.get("PARTITION_TRANSFORM_PROBE_MAX_VALUES", 1000)
+ if max_values and len(distinct) > max_values:
+ _count("declined.too_many_values")
+ logger.warning(
+ "A filter has %d distinct values, more than the %d a partition "
+ "transform probe evaluates; this query will not prune",
+ len(distinct),
+ max_values,
+ )
+ if errors is not None:
+ errors.append(
+ f"{len(distinct)} distinct values exceed "
+ f"PARTITION_TRANSFORM_PROBE_MAX_VALUES ({max_values})"
+ )
+ return None
+
# The value side of the premise `probe_sql_is_evaluable` checks on the
# finished statement: a value the driver and the engine would spell
# differently is not probed at all -- see `literal_round_trips`.
@@ -1844,6 +1883,7 @@ def evaluate_transform(
if not all(
literal_round_trips(value, dialect, database.backend) for value in distinct
):
+ _count("declined.value_does_not_round_trip")
logger.warning(
"A filter value is not written the same way by the driver and the "
"database, so the partition transform is not probed at it; this "
@@ -1851,17 +1891,9 @@ def evaluate_transform(
)
return None
- cache_key = _probe_cache_key(database, catalog, schema, transform, distinct)
- cached = _cache_get(cache_key)
+ cached = _cached_probe(database, catalog, schema, transform, distinct, errors)
if cached is None:
- cached = _run_probe(
- database, catalog, schema, transform, distinct, errors=errors
- )
- if cached is None:
- # Deliberately not cached: a transient engine blip would otherwise
- # keep the dataset pruning-free for the whole cache timeout.
- return None
- _cache_set(cache_key, cached)
+ return None
evaluated = dict(
zip((_hashable(value) for value in distinct), cached, strict=False)
@@ -1869,6 +1901,50 @@ def evaluate_transform(
return [evaluated[_hashable(value)] for value in values]
+def _cached_probe(
+ database: Database,
+ catalog: str | None,
+ schema: str | None,
+ transform: str,
+ distinct: list[Any],
+ errors: list[str] | None,
+) -> list[Any] | None:
+ """
+ `_run_probe` behind the probe cache, remembering failures as well.
+
+ A failure is remembered only briefly. Not at all, and a transform the
+ engine always rejects costs a warehouse round trip on every chart load for
+ as long as it is configured; for the full cache timeout, and one transient
+ engine blip keeps the dataset pruning-free for a day. The engine's own
+ messages are kept with it so the preview can still say why.
+ """
+ cache_key = _probe_cache_key(database, catalog, schema, transform, distinct)
+ cached = _cache_get(cache_key)
+ if cached is not None:
+ _count("cache_hit")
+ return cached
+
+ failure = _cache_get(_failure_cache_key(cache_key))
+ if failure is not None:
+ _count("failure_cache_hit")
+ if errors is not None:
+ errors.extend(failure)
+ return None
+
+ _count("cache_miss")
+ probe_errors: list[str] = []
+ cached = _run_probe(
+ database, catalog, schema, transform, distinct, errors=probe_errors
+ )
+ if errors is not None:
+ errors.extend(probe_errors)
+ if cached is None:
+ _cache_failure(cache_key, probe_errors)
+ return None
+ _cache_set(cache_key, cached)
+ return cached
+
+
def _run_probe(
database: Database,
catalog: str | None,
@@ -1924,6 +2000,7 @@ def _probe(
#
# Declining costs pruning, never correctness -- see `evaluate_transform`.
if reason := stored_expression_error(database, catalog, schema, transform):
+ _count("declined.unstorable_transform")
logger.warning(
"Refusing to probe a partition transform that is not a storable "
"expression; queries will not prune: %s",
@@ -1945,14 +2022,33 @@ def _probe(
# Declining rather than raising, for the same reason every other
# failure here declines: the chart query is already correct without
# the mirror, and it is the only thing the caller is waiting on.
+ _count("declined.unsafe_probe_sql")
logger.warning(
"Refusing to run a partition transform probe whose compiled SQL "
"is not a plain projection per input value; queries will not "
"prune"
)
return None
- frame = database.get_df(sql=sql, catalog=catalog, schema=schema)
+ # The chart query waits on this, and nothing else bounds it: chart
+ # requests carry no query deadline of their own. Best effort, like
+ # every other `utils.timeout` -- the alarm only fires on the main
+ # thread -- and the engine may go on running what it was abandoned on.
+ timeout = app.config.get("PARTITION_TRANSFORM_PROBE_TIMEOUT", 10)
+ started = time.monotonic()
+ try:
+ with (
+ utils.timeout(
+ seconds=timeout,
+ error_message=f"Partition transform probe exceeded {timeout}s",
+ )
+ if timeout and timeout > 0
+ else nullcontext()
+ ):
+ frame = database.get_df(sql=sql, catalog=catalog, schema=schema)
+ finally:
+ _timing("latency_ms", (time.monotonic() - started) * 1000)
if frame is None or frame.empty:
+ _count("error")
logger.warning(
"Partition transform probe returned no rows; skipping mirroring"
)
@@ -1969,6 +2065,7 @@ def _probe(
# than one that is merely short. `is_parseable` rejects that
# transform before it is ever stored; this is the backstop for one
# that predates the check.
+ _count("error")
logger.warning(
"Partition transform probe returned %d values for %d inputs",
column_count,
@@ -1981,10 +2078,19 @@ def _probe(
# above 2^53 comes back rounded. `_to_python_scalar` converts
# faithfully but runs after the loss, so the mirror asks for a key the
# warehouse does not hold and drops the row the filter matched.
- return [
+ values = [
_to_python_scalar(frame.iat[0, index]) for index in range(len(distinct))
]
+ _count("success")
+ return values
+ except SupersetTimeoutException as ex:
+ _count("timeout")
+ logger.warning("%s; queries will not prune", ex.message)
+ if errors is not None:
+ errors.append(ex.message)
+ return None
except Exception as ex: # pylint: disable=broad-except
+ _count("error")
logger.warning(
"Partition transform probe failed; queries will not prune",
exc_info=True,
@@ -2099,9 +2205,30 @@ def _probe_cache_key(
return f"partition_transform_probe:{digest}"
+def _failure_cache_key(cache_key: str) -> str:
+ """Where a failed probe for ``cache_key`` is remembered; see `_cache_failure`."""
+ return f"{cache_key}:failed"
+
+
+def _probe_cache() -> Any:
+ """
+ The first cache that keeps what it is given: ``CACHE_CONFIG``, then
+ ``DATA_CACHE_CONFIG``, then `_LOCAL_PROBE_CACHE`.
+
+ Both configured caches default to `NullCache`, under which every chart
+ query on a mapped dataset probed the warehouse again -- a successful probe
+ as much as a failed one -- and the verdict the pruning indicator reads was
+ never there to read.
+ """
+ for cache in (cache_manager.cache, cache_manager.data_cache):
+ if not isinstance(getattr(cache, "cache", cache), NullCache):
+ return cache
+ return _LOCAL_PROBE_CACHE
+
+
def _cache_get(key: str) -> list[Any] | None:
try:
- return cache_manager.cache.get(key)
+ return _probe_cache().get(key)
except Exception: # pylint: disable=broad-except # noqa: BLE001
return None
@@ -2109,11 +2236,43 @@ def _cache_get(key: str) -> list[Any] | None:
def _cache_set(key: str, value: list[Any]) -> None:
timeout = app.config.get("PARTITION_TRANSFORM_PROBE_CACHE_TIMEOUT", 24 * 60 * 60)
try:
- cache_manager.cache.set(key, value, timeout=timeout)
+ _probe_cache().set(key, value, timeout=timeout)
except Exception: # pylint: disable=broad-except
logger.warning("Could not cache partition transform probe", exc_info=True)
+def _cache_failure(cache_key: str, errors: list[str]) -> None:
+ """
+ Remember that the probe behind ``cache_key`` failed, with the engine's
+ messages, for ``PARTITION_TRANSFORM_PROBE_FAILURE_CACHE_TIMEOUT``.
+
+ Zero turns it off rather than meaning "forever", which is what a zero
+ timeout means to the cache backends.
+ """
+ timeout = app.config.get("PARTITION_TRANSFORM_PROBE_FAILURE_CACHE_TIMEOUT", 300)
+ if not timeout or timeout <= 0:
+ return
+ try:
+ _probe_cache().set(_failure_cache_key(cache_key), errors, timeout=timeout)
+ except Exception: # pylint: disable=broad-except
+ logger.warning("Could not cache partition transform failure", exc_info=True)
+
+
+def _count(outcome: str) -> None:
+ # Metrics are advisory; a broken statsd client must not cost the chart.
+ try:
+ stats_logger_manager.instance.incr(f"{PROBE_METRIC_PREFIX}.{outcome}")
+ except Exception: # pylint: disable=broad-except # noqa: BLE001
+ logger.debug("Could not emit partition probe metric", exc_info=True)
+
+
+def _timing(name: str, value: float) -> None:
+ try:
+ stats_logger_manager.instance.timing(f"{PROBE_METRIC_PREFIX}.{name}", value)
+ except Exception: # pylint: disable=broad-except # noqa: BLE001
+ logger.debug("Could not emit partition probe metric", exc_info=True)
+
+
def _mirror_verdict_cache_key(
database: Database,
catalog: str | None,
@@ -2160,11 +2319,11 @@ def record_mirror_verdict(
"""
Remember whether this transform produced a mirror, for the UI to read.
- Advisory only. Nothing in the query path consults it: a probe that failed
- once is retried on the next chart exactly as it is today, because declining
- to prune for a whole cache timeout over one transient engine blip is the
- trade `evaluate_transform` deliberately refuses -- see the comment where it
- declines to cache a failure.
+ Advisory only. Nothing in the query path consults it: a failed probe is
+ remembered for ``PARTITION_TRANSFORM_PROBE_FAILURE_CACHE_TIMEOUT`` and then
+ retried, because declining to prune for a whole cache timeout over one
+ transient engine blip is the trade `evaluate_transform` deliberately
+ refuses -- see the comment where it caches a failure.
What this is for is the *claim*. `is_transform_active` parses the transform
and no more, so a transform the database rejects -- a misspelled function
@@ -2179,7 +2338,7 @@ def record_mirror_verdict(
"""
timeout = app.config.get("PARTITION_TRANSFORM_PROBE_CACHE_TIMEOUT", 24 * 60 * 60)
try:
- cache_manager.cache.set(
+ _probe_cache().set(
_mirror_verdict_cache_key(database, catalog, schema, transform),
mirrors,
timeout=timeout,
@@ -2201,11 +2360,11 @@ def known_mirror_verdict(
reporting a working mapping as broken until something probes it would
trade one wrong claim for another. The alternative -- probing from the
read side -- would put a warehouse round trip on every Explore and
- dashboard load, and a failing transform is not cached at all, so it would
- be a round trip every single time.
+ dashboard load, and a failing transform is only cached briefly, so it
+ would be a round trip nearly every time.
"""
try:
- verdict = cache_manager.cache.get(
+ verdict = _probe_cache().get(
_mirror_verdict_cache_key(database, catalog, schema, transform)
)
except Exception: # pylint: disable=broad-except # noqa: BLE001
diff --git a/tests/unit_tests/conftest.py b/tests/unit_tests/conftest.py
index 07cf4e3193d..93d58a81c06 100644
--- a/tests/unit_tests/conftest.py
+++ b/tests/unit_tests/conftest.py
@@ -215,6 +215,20 @@ def app_context(app: SupersetApp) -> Iterator[None]:
yield
+@pytest.fixture(autouse=True)
+def _clear_local_partition_probe_cache() -> Iterator[None]:
+ """
+ The test app runs a null cache, so partition probes fall back to a
+ per-process one -- which would otherwise serve one test's mocked probe
+ result to the next test that happens to share its cache key.
+ """
+ from superset.connectors.sqla.partition_mapping import _LOCAL_PROBE_CACHE
+
+ _LOCAL_PROBE_CACHE.clear()
+ yield
+ _LOCAL_PROBE_CACHE.clear()
+
+
@pytest.fixture
def full_api_access(mocker: MockerFixture) -> Union[Iterator[None], None]:
"""
diff --git a/tests/unit_tests/connectors/sqla/partition_mapping_test.py b/tests/unit_tests/connectors/sqla/partition_mapping_test.py
index e69a5995007..e49c21bb6f0 100644
--- a/tests/unit_tests/connectors/sqla/partition_mapping_test.py
+++ b/tests/unit_tests/connectors/sqla/partition_mapping_test.py
@@ -18,6 +18,7 @@
from __future__ import annotations
+from contextlib import nullcontext
from datetime import datetime
from decimal import Decimal
from importlib import import_module
@@ -2112,8 +2113,40 @@ def test_the_cache_key_includes_the_catalog_and_schema(app: Flask) -> None:
assert _probe(database).call_count == 2
-def test_a_failed_probe_is_not_cached(app: Flask) -> None:
- """Caching a failure would keep a transient engine blip pruning-free for a day."""
+def test_a_failed_probe_is_remembered_briefly(app: Flask) -> None:
+ """
+ A transform the engine always rejects must not cost a warehouse round trip
+ on every chart load -- and the preview must still be able to say why.
+ """
+ database = Database(database_name="probe_db", sqlalchemy_uri="sqlite://")
+ database.get_df = MagicMock( # type: ignore[method-assign]
+ side_effect=RuntimeError("function no_such_fn does not exist")
+ )
+ errors: list[str] = []
+
+ with app.app_context():
+ assert evaluate_transform(database, None, None, "lower(:value)", ["x"]) is None
+ assert (
+ evaluate_transform(
+ database, None, None, "lower(:value)", ["x"], errors=errors
+ )
+ is None
+ )
+
+ _probe(database).assert_called_once()
+ assert errors == ["function no_such_fn does not exist"]
+
+
+def test_a_failed_probe_is_retried_once_its_failure_expires(
+ app: Flask, mocker: MockerFixture
+) -> None:
+ """
+ Remembering a failure for the full cache timeout would keep a transient
+ engine blip pruning-free for a day; zero turns the failure cache off.
+ """
+ mocker.patch.dict(
+ app.config, {"PARTITION_TRANSFORM_PROBE_FAILURE_CACHE_TIMEOUT": 0}
+ )
database = Database(database_name="probe_db", sqlalchemy_uri="sqlite://")
database.get_df = MagicMock( # type: ignore[method-assign]
side_effect=[
@@ -2127,6 +2160,135 @@ def test_a_failed_probe_is_not_cached(app: Flask) -> None:
assert evaluate_transform(database, None, None, "lower(:value)", ["x"]) == [42]
+def test_a_filter_wider_than_the_probe_cap_is_not_probed(
+ app: Flask, mocker: MockerFixture
+) -> None:
+ """
+ Every value is one select column, and PostgreSQL refuses a 1,665th: a wide
+ `IN` list would otherwise fail on the engine on every chart load.
+ """
+ mocker.patch.dict(app.config, {"PARTITION_TRANSFORM_PROBE_MAX_VALUES": 3})
+ database = _database_returning([1, 2, 3])
+ errors: list[str] = []
+
+ with app.app_context():
+ assert (
+ evaluate_transform(
+ database,
+ None,
+ None,
+ "lower(:value)",
+ ["a", "b", "c", "d"],
+ errors=errors,
+ )
+ is None
+ )
+ # The cap counts distinct values, so duplicates do not trip it.
+ assert evaluate_transform(
+ database, None, None, "lower(:value)", ["a", "b", "c", "a"]
+ ) == [1, 2, 3, 1]
+
+ _probe(database).assert_called_once()
+ assert errors == [
+ "4 distinct values exceed PARTITION_TRANSFORM_PROBE_MAX_VALUES (3)"
+ ]
+
+
+def test_a_probe_runs_under_its_own_timeout(app: Flask, mocker: MockerFixture) -> None:
+ """Chart requests carry no deadline, so nothing else bounds the probe."""
+ from superset.errors import ErrorLevel, SupersetErrorType
+ from superset.exceptions import SupersetTimeoutException
+
+ mocker.patch.dict(app.config, {"PARTITION_TRANSFORM_PROBE_TIMEOUT": 7})
+ timeout = mocker.patch(
+ "superset.connectors.sqla.partition_mapping.utils.timeout",
+ return_value=nullcontext(),
+ )
+ database = Database(database_name="probe_db", sqlalchemy_uri="sqlite://")
+ database.get_df = MagicMock( # type: ignore[method-assign]
+ side_effect=SupersetTimeoutException(
+ error_type=SupersetErrorType.BACKEND_TIMEOUT_ERROR,
+ message="Partition transform probe exceeded 7s",
+ level=ErrorLevel.ERROR,
+ )
+ )
+ errors: list[str] = []
+
+ with app.app_context():
+ assert (
+ evaluate_transform(
+ database, None, None, "lower(:value)", ["x"], errors=errors
+ )
+ is None
+ )
+
+ assert timeout.call_args.kwargs["seconds"] == 7
+ assert errors == ["Partition transform probe exceeded 7s"]
+
+
+def test_a_null_cache_falls_back_to_a_local_one(app: Flask) -> None:
+ """
+ `CACHE_CONFIG` and `DATA_CACHE_CONFIG` both default to `NullCache`, under
+ which every chart query on a mapped dataset probed again.
+ """
+ from flask_caching import Cache
+
+ from superset.extensions import cache_manager
+
+ null = Cache(config={"CACHE_TYPE": "NullCache"})
+ null.init_app(app)
+ database = _database_returning([1767225600])
+
+ with patch.object(cache_manager, "_cache", null):
+ with patch.object(cache_manager, "_data_cache", null):
+ with app.app_context():
+ for _ in range(2):
+ assert evaluate_transform(
+ database, None, None, "unix_timestamp(:value)", ["x"]
+ ) == [1767225600]
+ assert (
+ known_mirror_verdict(database, None, None, "unix_timestamp(:value)")
+ is True
+ )
+
+ _probe(database).assert_called_once()
+
+
+def test_the_probe_reports_its_outcome_and_latency(
+ app: Flask, mocker: MockerFixture
+) -> None:
+ stats = mocker.patch(
+ "superset.connectors.sqla.partition_mapping.stats_logger_manager"
+ ).instance
+ database = _database_returning([1])
+
+ with app.app_context():
+ evaluate_transform(database, None, None, "lower(:value)", ["x"])
+ evaluate_transform(database, None, None, "lower(:value)", ["x"])
+
+ assert [call.args[0] for call in stats.incr.call_args_list] == [
+ "partition_mapping.probe.cache_miss",
+ "partition_mapping.probe.success",
+ "partition_mapping.probe.cache_hit",
+ ]
+ stats.timing.assert_called_once()
+ assert stats.timing.call_args.args[0] == "partition_mapping.probe.latency_ms"
+
+
+def test_a_broken_stats_logger_does_not_cost_the_probe(
+ app: Flask, mocker: MockerFixture
+) -> None:
+ stats = mocker.patch(
+ "superset.connectors.sqla.partition_mapping.stats_logger_manager"
+ ).instance
+ stats.incr.side_effect = RuntimeError("statsd down")
+ stats.timing.side_effect = RuntimeError("statsd down")
+ database = _database_returning([1])
+
+ with app.app_context():
+ assert evaluate_transform(database, None, None, "lower(:value)", ["x"]) == [1]
+
+
def test_a_null_monotonic_flag_reads_as_not_declared(app: Flask) -> None:
"""
The column is nullable because the legacy datasource editor writes NULL for