fix(embedded): restore chart rendering for guest dashboards

Embedded dashboards stopped rendering their charts for guest (embedded)
users: the dashboard payload dropped member charts and withheld their
form_data, so the frontend had nothing to render and hung on the loading
spinner.

An embedded guest carries no standalone datasource grant, so the chart
branch of raise_for_access could not recognise a guest during dashboard
serialization (the dashboard/form_data context a chart-data request
provides is absent there). Grant a guest access to a member chart of a
dashboard their token covers, bounded by the token's dataset allowlist.
Non-guest access is unchanged.

Also keep the embedded Playwright helper's error-surfacing diagnostic and
the cold-CI render timeout, which is what pinned this down.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.github/CODEOWNERS

@@ -1,22 +1,3 @@
-# Notify all committers of DB migration changes, per SIP-59
-
-/superset/migrations/ @mistercrunch @michael-s-molina @betodealmeida @eschutho @sadpandajoe @rusackas
-
-# Notify some committers of changes in the components
-
-/superset-frontend/src/components/Select/ @michael-s-molina @geido @kgabryje
-/superset-frontend/src/components/MetadataBar/ @michael-s-molina @geido @kgabryje
-/superset-frontend/src/components/DropdownContainer/ @michael-s-molina @geido @kgabryje
-
-# Notify Helm Chart maintainers about changes in it
-
-/helm/superset/ @dpgaspar @villebro @nytai @michael-s-molina @mistercrunch @rusackas @Antonio-RiveroMartnez @hainenber
-
-# Notify E2E test maintainers of changes
-
-/superset-frontend/playwright/ @sadpandajoe @geido @eschutho @rusackas @mistercrunch
-/superset-frontend/cypress-base/ @sadpandajoe @geido @eschutho @rusackas @mistercrunch
-
 # Notify PMC members of changes to GitHub Actions
 
 /.github/ @villebro @geido @eschutho @rusackas @betodealmeida @nytai @mistercrunch @kgabryje @sha174n @dpgaspar @sadpandajoe @hainenber
@@ -30,16 +11,12 @@
 
 /.asf.yaml @villebro @geido @eschutho @rusackas @betodealmeida @nytai @mistercrunch @kgabryje @dpgaspar @sha174n @Antonio-RiveroMartnez
 
-# Maps are a finicky contribution process we care about
+# Maps are fragile and political. GeoJson edits MUST be made in the Jupyter notebook or they'll be overwritten.
 
 **/*.geojson @villebro @rusackas
 **/*.ipynb @villebro @rusackas
 /superset-frontend/plugins/plugin-chart-country-map/ @villebro @rusackas
 
-# Notify translation maintainers of changes to translations
-
-/superset/translations/ @sfirke @rusackas @villebro @sadpandajoe @hainenber
-
 # Notify PMC members of changes to extension-related files
 
 /docs/developer_portal/extensions/ @michael-s-molina @villebro @rusackas

.github/workflows/generate-FOSSA-report.yml

@@ -37,7 +37,7 @@ jobs:
           persist-credentials: false
           submodules: recursive
       - name: Setup Java
-        uses: actions/setup-java@b6effb05e454b25005698d916606bdc6ffcbf961 # v5.7.0
+        uses: actions/setup-java@03ad4de0992f5dab5e18fcb136590ce7c4a0ac95 # v5.6.0
         with:
           distribution: "temurin"
           java-version: "11"

.github/workflows/license-check.yml

@@ -23,7 +23,7 @@ jobs:
           persist-credentials: false
           submodules: recursive
       - name: Setup Java
-        uses: actions/setup-java@b6effb05e454b25005698d916606bdc6ffcbf961 # v5.7.0
+        uses: actions/setup-java@03ad4de0992f5dab5e18fcb136590ce7c4a0ac95 # v5.6.0
         with:
           distribution: "temurin"
           java-version: "11"

.github/workflows/superset-docs-deploy.yml

@@ -76,7 +76,7 @@ jobs:
           node-version-file: "./docs/.nvmrc"
       - name: Setup Python
         uses: ./.github/actions/setup-backend/
-      - uses: actions/setup-java@b6effb05e454b25005698d916606bdc6ffcbf961 # v5.7.0
+      - uses: actions/setup-java@03ad4de0992f5dab5e18fcb136590ce7c4a0ac95 # v5.6.0
         with:
           distribution: "zulu"
           java-version: "21"

UPDATING.md

@@ -54,6 +54,43 @@ pip install playwright && playwright install chromium
 2. Remove any references to the removed config keys from custom `superset_config.py`
 3. If you subclassed `MachineAuthProvider`, remove any `authenticate_webdriver` override and migrate auth logic to `authenticate_browser_context`
 
+### CSV/XLSX report exports of Table charts keep raw numeric values
+
+Table and Pivot Table charts sent as text in a report email now apply the
+chart's number and currency formatting so the values match what a user sees in
+Explore. As part of this, the CSV and XLSX result formats return early before
+formatting: previously the Table post-processor applied `d3NumberFormat` to
+every result format, so CSV/XLSX exports contained pre-formatted strings.
+CSV/XLSX exports now preserve numeric values and column types, which is better
+for downstream analysis but is a visible change for anyone who relied on the
+formatted text in those files. The rendered email body (the only place the
+formatting is intended for) is unaffected.
+
+### SQLAlchemy bumped to 2.0, flask-sqlalchemy to 3.1.1
+
+Superset's core ORM dependencies move from SQLAlchemy 1.4 to 2.0 and
+flask-sqlalchemy `<3.0` to 3.1.1, completing the migration tracked in
+[discussion #40273](https://github.com/apache/superset/discussions/40273).
+
+**Custom `db_engine_specs`, plugins, or extensions that import SQLAlchemy
+internals directly** should review the
+[SQLAlchemy 1.4-to-2.0 migration guide](https://docs.sqlalchemy.org/en/20/changelog/migration_20.html)
+for API changes that affect them — most 1.4 code already runs unmodified
+under 2.0's compatibility mode, but patterns like `Engine.execute()`,
+string-keyed `Row` access, and `MetaData(bind=)` are removed outright.
+
+**Several optional DB-connector extras remain capped below their
+SQLAlchemy-2.0-only releases**, either because that bump is a separate
+follow-up ([#42891](https://github.com/apache/superset/pull/42891): dremio,
+exasol, firebird, redshift, risingwave) or because the upstream dialect
+package has no SQLAlchemy 2.0 support yet at all (aurora-data-api, d1,
+kusto, solr; ocient's 2.0 compatibility is unverified). Installing one of
+these extras continues to pull a SQLAlchemy-1.4-line version of that
+dialect; each package's constraint in `pyproject.toml` documents why.
+
+No application-level configuration changes are required for deployments
+that don't touch SQLAlchemy directly.
+
 ### Soft delete is on by default, and purging is live
 
 `SOFT_DELETE` now ships **on** (`DEFAULT_FEATURE_FLAGS`), so deleting a
@@ -159,6 +196,36 @@ will now get a TypeScript error and must remove the prop; keeping a manual
 override was exactly the footgun this change removes (see #42510). No
 callers in the Superset frontend codebase itself passed this prop.
 
+### Row-level security now filters table reads a same-named CTE used to hide
+
+`extract_tables_from_statement()` decided whether a reference was a CTE by matching its
+bare name against the enclosing scope's CTE names; it now resolves the name through
+`Scope.cte_sources`. Three kinds of real table read whose bare name collided with a CTE's
+were mistaken for the CTE and dropped from a statement's tables, so they were neither
+RLS-filtered nor access-checked: a schema- or catalog-qualified reference, a non-recursive
+CTE's own name inside its body, and a forward reference to a later `WITH` item.
+
+```sql
+WITH orders AS (SELECT 1 AS d) SELECT * FROM (SELECT * FROM public.orders) AS z
+WITH orders AS (SELECT * FROM orders) SELECT * FROM orders
+WITH q1 AS (SELECT key FROM q2), q2 AS (SELECT 1 AS key) SELECT * FROM q1
+```
+
+Each read is now reported, so it is filtered when `RLS_IN_SQLLAB` is enabled, matched
+against `DISALLOWED_SQL_TABLES`, and requires dataset access under
+`raise_for_access(force_dataset_match=True)`. A query that previously ran, reading those
+rows unfiltered, may now be filtered or rejected. There is no opt-out — the previous
+behavior was a row-level-security bypass.
+
+### Table aliases keep their quoting through the row-level security rewrite
+
+Both RLS transformers took the table alias as a string with its quoting stripped and
+emitted it verbatim; they now carry the parsed identifier. Emitted SQL is unchanged for an
+unquoted identifier; a quoted one keeps its quoting, and a column-alias list
+(`FROM t AS x (c1, c2)`) survives the rewrite instead of being dropped. This repairs
+row-level security for any aliased table on Snowflake, and for at least one statement shape
+on MSSQL where the rewrite previously raised `AttributeError`.
+
 ### Principal listing APIs now honour related-field filters
 
 Two authorization-related listing behaviors changed for API clients. Neither
@@ -828,6 +895,8 @@ With the flag enabled: `DELETE /api/v1/chart/<id>` no longer hard-deletes the ch
 
 - [39914](https://github.com/apache/superset/pull/39914) `ALERT_REPORT_SLACK_V2` now defaults to `True` and the legacy Slack v1 integration (`Slack` recipient type, `files.upload` API) is deprecated for removal in the next major. Slack blocked new apps from `files.upload` in May 2024 and fully retired the method for all apps on November 12, 2025; because the v1 path sends files through `files.upload`, v1 file-bearing sends now fail at the API level — only text-only `chat_postMessage` still works via the legacy path. Grant your Slack bot the `channels:read` and `groups:read` scopes so existing `Slack` recipients can be auto-upgraded to `SlackV2` on next send. Operators who explicitly override the flag to `False`, or whose Slack bot is missing those scopes, will see deprecation warnings while text-only sends continue through the legacy path.
 
+- [42089](https://github.com/apache/superset/pull/42089) automatically upgrades resolvable Slack v1 recipients, preserves text-only v1 delivery with execution warnings when migration cannot finish, and rejects retired v1 file uploads with actionable scope guidance. Slack delivery uses at-most-once terminal writes and a schedule-wide retry budget configured by `SLACK_SEND_RETRY_MAX_TIME`, clamped to the report's remaining working timeout. Deployments using `SupersetMetastoreCache` for the Slack channel cache must schedule the `slack.cache_channels` Celery task to repopulate misses outside report transactions; see [Alerts and Reports](https://superset.apache.org/admin-docs/configuration/alerts-reports#slack-delivery-timeouts-and-retries).
+
 ### Soft delete and restore for dashboards
 
 **Everything in this section applies only when the `SOFT_DELETE` feature flag is enabled. The flag defaults to `False`** (`@lifecycle: development`), so on a default deployment `DELETE /api/v1/dashboard/<id>` continues to **hard-delete permanently** — nothing is recoverable. Enable `SOFT_DELETE` to get the behavior described below.

docs/admin_docs/configuration/alerts-reports.mdx

@@ -83,6 +83,28 @@ SLACK_CACHE_TIMEOUT = int(timedelta(days=2).total_seconds())
 SLACK_API_RATE_LIMIT_RETRY_COUNT = 5
 ```
 
+When the cache backend is `SupersetMetastoreCache`, report execution does not
+write channel listings into the cache because that backend commits the report's
+database session. Schedule the dedicated warm-up task so cache misses are
+repopulated outside report transactions:
+
+```python
+from celery.schedules import crontab
+
+from superset.config import CeleryConfig
+
+class CustomCeleryConfig(CeleryConfig):
+    beat_schedule = {
+        **CeleryConfig.beat_schedule,
+        "slack.cache_channels": {
+            "task": "slack.cache_channels",
+            "schedule": crontab(minute="0", hour="*"),
+        },
+    }
+
+CELERY_CONFIG = CustomCeleryConfig
+```
+
 #### Slack Enterprise Grid (org-scoped tokens)
 
 On a Slack Enterprise Grid org, an org-scoped token spans multiple workspaces, so
@@ -98,6 +120,38 @@ SLACK_TEAM_ID = "T01234567"
 This defaults to `None` and only needs to be set when using an org-scoped token;
 it is accepted but ignored for standard workspace-level tokens.
 
+#### Slack delivery timeouts and retries
+
+Slack delivery uses a request timeout and an application retry budget:
+
+```python
+# Timeout for one Slack API request, in seconds
+SLACK_API_TIMEOUT = 30
+
+# Retry budget shared by every Slack destination and upload phase
+SLACK_SEND_RETRY_MAX_TIME = 150
+
+# Number of explicit HTTP 429 responses retried using Slack's Retry-After value
+SLACK_API_RATE_LIMIT_RETRY_COUNT = 2
+
+# Cooldown after an on-demand channel-cache refresh
+SLACK_CHANNEL_REFRESH_COOLDOWN_SECONDS = 300
+```
+
+All channels and upload phases in one report execution share a single
+`SLACK_SEND_RETRY_MAX_TIME` budget. This prevents a large recipient list from
+multiplying the report's wall-clock retry time. The budget is also clamped to
+the report's remaining working timeout, leaving Celery's configured timeout lag
+available for final state persistence. The effective configured budget is at
+least one second longer than `SLACK_API_TIMEOUT`.
+
+To avoid posting the same report twice, Superset does not replay terminal
+`chat.postMessage` or `files.completeUploadExternal` operations after ambiguous
+server or transport failures. Explicit Slack HTTP 429 responses remain
+retryable. These delivery settings and semantics apply to Slack v2 reports and
+legacy text-only Slack delivery, independently of the
+`ALERT_REPORT_SLACK_V2` feature flag.
+
 ### Webhook integration
 
 Superset can send alert and report notifications to any HTTP endpoint — useful for chat platforms, incident management tools, or custom automation.

pyproject.toml

@@ -60,15 +60,11 @@ dependencies = [
     "flask-login>=0.6.0, < 1.0",
     "flask-migrate>=4.1.0, <5.0",
     "flask-session>=0.4.0, <1.0",
-    # Pinned explicitly below 3.0: 3.0.5 resolves without conflict and
-    # supports both SQLAlchemy 1.4 and 2.0, but real CI runs surfaced a
-    # structural incompatibility with Superset's current session/app-context
-    # handling across Celery task boundaries (see PR #42542) -- widespread
-    # "NoneType has no attribute X" failures and MySQL lock-wait timeouts,
-    # not just a connection-pool quirk. Needs dedicated investigation, not a
-    # driver-compat-prep bump; revisit alongside the actual SQLAlchemy 2.0
-    # core bump (discussion #40273, step 6).
-    "flask-sqlalchemy>=2.5.1, <4.0",
+    # Bumped to 3.1.1 alongside the SQLAlchemy 2.0 core bump (discussion
+    # #40273, step 6), which resolves the session/app-context handling
+    # across Celery task boundaries that previously blocked this (see
+    # PR #42542).
+    "flask-sqlalchemy>=3.1.1, <4.0",
     "flask-wtf>=1.3.0, <2.0",
     "geopy",
     "greenlet<=3.5.4, >=3.5.4",
@@ -115,7 +111,7 @@ dependencies = [
     "sshtunnel>=0.4.0, <0.5",
     "simplejson>=4.1.1",
     "slack_sdk>=3.43.0, <4",
-    "sqlalchemy>=1.4.43, <2", # 1.4.43 adds the python-oracledb (oracle+oracledb) dialect
+    "sqlalchemy>=2.0.0, <2.1",
     "sqlalchemy-continuum>=1.6.0, <2.0.0",
     "sqlalchemy-utils>=0.42.1, <0.43", # expanding lowerbound to work with pydoris
     "sqlglot>=30.16.0, <31", # 30.16.0 adds Trino inline UDF IF/CASE routine statement parsing
@@ -164,11 +160,10 @@ databricks = [
 datafusion = ["flightsql-dbapi>=0.2.2, <0.3"]
 db2 = ["ibm-db-sa<=0.4.4, >=0.4.4"]
 denodo = ["denodo-sqlalchemy>=2.0.5,<2.1.0"]
-# sqlalchemy-dremio 3.0.5+ hard-pins sqlalchemy~=2.0.41, dropping 1.4; 3.0.4
-# is the last dual-compat release. Capped below 3.0.5 for now; widen back to
-# <4 in lockstep with Superset's own SQLAlchemy 2.0 core bump (discussion
-# #40273), not before.
-dremio = ["sqlalchemy-dremio>=1.2.1, <3.0.5"]
+# sqlalchemy-dremio 3.0.5+ hard-pins sqlalchemy~=2.0.41, dropping 1.4.
+# Widened now that Superset's own SQLAlchemy 2.0 core bump has landed
+# (discussion #40273).
+dremio = ["sqlalchemy-dremio>=3.0.5, <4"]
 # <2 was an artificial ceiling; upstream has no SQLAlchemy version cap and
 # 1.1.10 already supports SQLAlchemy 2.0 (added `import_dbapi` in 1.1.7).
 drill = ["sqlalchemy-drill>=1.1.10, <3"]
@@ -181,10 +176,9 @@ dynamodb = ["pydynamodb>=0.8.2"]
 solr = ["sqlalchemy-solr>=0.2.4.3"]
 elasticsearch = ["elasticsearch-dbapi>=0.2.13, <0.3.0"]
 # sqlalchemy-exasol cuts hard from SQLAlchemy 1.4-only (<6.0.0) to 2.0-only
-# (>=6.0.0) with no dual-compat release. Capped below 6.0.0 for now; bump to
-# >=6.0.0,<8.0 in lockstep with Superset's own SQLAlchemy 2.0 core bump
-# (discussion #40273), not before.
-exasol = ["sqlalchemy-exasol>=2.4.0, <6.0.0"]
+# (>=6.0.0) with no dual-compat release. Bumped now that Superset's own
+# SQLAlchemy 2.0 core bump has landed (discussion #40273).
+exasol = ["sqlalchemy-exasol>=6.0.0, <8.0"]
 excel = ["xlrd>=2.0.2, <2.1"]
 # Async dashboard "Export Data/Images to Excel": uploads the workbook to S3 and
 # emails a pre-signed link. boto3 is imported lazily by superset.utils.s3, so
@@ -199,9 +193,9 @@ fastmcp = [
 ]
 # sqlalchemy-firebird >=2.0.0 unconditionally requires SQLAlchemy 2.0 on
 # Python >=3.8 (which covers Superset's >=3.11 floor), with no dual-compat
-# release. Capped below 2.0.0 for now; bump to >=2.2.0 in lockstep with
-# Superset's own SQLAlchemy 2.0 core bump (discussion #40273), not before.
-firebird = ["sqlalchemy-firebird>=0.8.0, <2.0.0"]
+# release. Bumped now that Superset's own SQLAlchemy 2.0 core bump has
+# landed (discussion #40273).
+firebird = ["sqlalchemy-firebird>=2.2.0"]
 firebolt = ["firebolt-sqlalchemy>=1.1.2, <2"]
 gevent = ["gevent>=26.7.0"]
 gsheets = ["shillelagh[gsheetsapi]>=1.4.5, <2"]
@@ -240,15 +234,14 @@ presto = ["pyhive[presto]>=0.6.5"]
 trino = ["trino>=0.338.0"]
 prophet = ["prophet>=1.3.0, <2"]
 # sqlalchemy-redshift cuts hard from SQLAlchemy 1.4-only (0.8.x) to 2.0-only
-# (>=1.0.0) with no dual-compat release; the existing <0.9 ceiling already
-# keeps this on the 1.4-only line. Bump to >=1.0.0 in lockstep with
-# Superset's own SQLAlchemy 2.0 core bump (discussion #40273), not before.
-redshift = ["sqlalchemy-redshift>=0.8.1, <0.9"]
+# (>=1.0.0) with no dual-compat release. Bumped now that Superset's own
+# SQLAlchemy 2.0 core bump has landed (discussion #40273).
+redshift = ["sqlalchemy-redshift>=1.0.0"]
 # No release of sqlalchemy-risingwave has ever supported both SQLAlchemy 1.4
-# and 2.0 (version numbers don't track SQLAlchemy compat monotonically); pin
-# to the newest 1.4-only release for now. Bump to >=2.0.0 in lockstep with
-# Superset's own SQLAlchemy 2.0 core bump (discussion #40273), not before.
-risingwave = ["sqlalchemy-risingwave>=1.4.1, <3.0.0"]
+# and 2.0 (version numbers don't track SQLAlchemy compat monotonically).
+# Bumped to the 2.0-only line now that Superset's own SQLAlchemy 2.0 core
+# bump has landed (discussion #40273).
+risingwave = ["sqlalchemy-risingwave>=2.0.0"]
 shillelagh = ["shillelagh[all]>=1.4.5, <2"]
 singlestore = ["sqlalchemy-singlestoredb>=1.2.1, <2"]
 snowflake = ["snowflake-sqlalchemy>=1.11.0, <2"]

pytest.ini

@@ -23,25 +23,5 @@ python_files = *_test.py test_*.py *_tests.py *viz/utils.py
 asyncio_mode = auto
 
 # `ignore` is effectively equivalent to `-p no:warnings`.
-# Always print RemovedIn20Warning when SQLALCHEMY_WARN_20=1.
-# Additionally, raise errors for refactored RemovedIn20Warning cases to prevent regression.
 filterwarnings =
     ignore
-    always::sqlalchemy.exc.RemovedIn20Warning
-    error:Passing a string to Connection.execute\(\) is deprecated:sqlalchemy.exc.RemovedIn20Warning
-    error:"Query" object is being merged into a Session:sqlalchemy.exc.RemovedIn20Warning
-    error:"ReportExecutionLog" object is being merged into a Session:sqlalchemy.exc.RemovedIn20Warning
-    error:"ReportRecipients" object is being merged into a Session:sqlalchemy.exc.RemovedIn20Warning
-    error:"SavedQuery" object is being merged into a Session:sqlalchemy.exc.RemovedIn20Warning
-    error:"SqlaTable" object is being merged into a Session:sqlalchemy.exc.RemovedIn20Warning
-    error:"SqlMetric" object is being merged into a Session:sqlalchemy.exc.RemovedIn20Warning
-    error:"SSHTunnel" object is being merged into a Session:sqlalchemy.exc.RemovedIn20Warning
-    error:"TableColumn" object is being merged into a Session:sqlalchemy.exc.RemovedIn20Warning
-    error:"TaggedObject" object is being merged into a Session:sqlalchemy.exc.RemovedIn20Warning
-    error:The autoload parameter is deprecated:sqlalchemy.exc.RemovedIn20Warning
-    error:The connection.execute\(\) method:sqlalchemy.exc.RemovedIn20Warning
-    error:The current statement is being autocommitted using implicit autocommit:sqlalchemy.exc.RemovedIn20Warning
-    error:The ``declarative_base\(\)`` function is now available:sqlalchemy.exc.RemovedIn20Warning
-    error:The Engine.execute\(\) method is considered legacy:sqlalchemy.exc.RemovedIn20Warning
-    error:The legacy calling style of select\(\) is deprecated:sqlalchemy.exc.RemovedIn20Warning
-    error:The "whens" argument to case:sqlalchemy.exc.RemovedIn20Warning

requirements/base.txt

@@ -144,7 +144,7 @@ flask-migrate==4.1.0
     # via apache-superset (pyproject.toml)
 flask-session==0.8.0
     # via apache-superset (pyproject.toml)
-flask-sqlalchemy==2.5.1
+flask-sqlalchemy==3.1.1
     # via
     #   apache-superset (pyproject.toml)
     #   flask-appbuilder
@@ -381,7 +381,7 @@ six==1.17.0
     #   wtforms-json
 slack-sdk==3.43.0
     # via apache-superset (pyproject.toml)
-sqlalchemy==1.4.54
+sqlalchemy==2.0.51
     # via
     #   apache-superset (pyproject.toml)
     #   alembic
@@ -419,6 +419,7 @@ typing-extensions==4.16.0
     #   pyopenssl
     #   referencing
     #   shillelagh
+    #   sqlalchemy
     #   typing-inspection
 typing-inspection==0.4.2
     # via pydantic

requirements/development.txt

@@ -306,7 +306,7 @@ flask-session==0.8.0
     # via
     #   -c requirements/base-constraint.txt
     #   apache-superset
-flask-sqlalchemy==2.5.1
+flask-sqlalchemy==3.1.1
     # via
     #   -c requirements/base-constraint.txt
     #   apache-superset
@@ -950,7 +950,7 @@ slack-sdk==3.43.0
     #   apache-superset
 sniffio==1.3.1
     # via anyio
-sqlalchemy==1.4.54
+sqlalchemy==2.0.51
     # via
     #   -c requirements/base-constraint.txt
     #   alembic
@@ -1033,6 +1033,7 @@ typing-extensions==4.16.0
     #   pyopenssl
     #   referencing
     #   shillelagh
+    #   sqlalchemy
     #   starlette
     #   typing-inspection
 typing-inspection==0.4.2

superset-core/pyproject.toml

@@ -45,7 +45,7 @@ dependencies = [
     "isodate>=0.7.0",
     "pyarrow>=16.0.0",
     "pydantic>=2.8.0",
-    "sqlalchemy>=1.4.0,<2.0",
+    "sqlalchemy>=2.0.0,<2.1",
     "sqlalchemy-utils>=0.38.0, <0.43", # expanding lowerbound to work with pydoris
     "sqlglot>=30.8.0, <31",
     "typing-extensions>=4.0.0",

superset-frontend/package-lock.json

@@ -272,7 +272,7 @@
         "swc-loader": "^0.2.7",
         "ts-jest": "^29.4.12",
         "tscw-config": "^1.1.2",
-        "tsx": "^4.23.5",
+        "tsx": "^4.23.7",
         "typescript": "5.4.5",
         "unzipper": "^0.12.5",
         "wait-on": "^9.1.0",
@@ -39964,9 +39964,9 @@
       "license": "0BSD"
     },
     "node_modules/tsx": {
-      "version": "4.23.5",
-      "resolved": "https://registry.npmjs.org/tsx/-/tsx-4.23.5.tgz",
-      "integrity": "sha512-rw55FUaqOoI7RvlQwLbhO4nSDApnQ4/CykPuiQ/EPvtrX3WA9Ig55jIt9VvbBJbzJuj12ueRu4PMZ2SxPVbihg==",
+      "version": "4.23.7",
+      "resolved": "https://registry.npmjs.org/tsx/-/tsx-4.23.7.tgz",
+      "integrity": "sha512-3f/u/+UDCNQ7iwUZW9FCMnNGIHzElGJYh0S/yy8IvWSsn5O7fEO/897FaG7FA2W8yryiRyuwXZ1PYLAKYaqSuQ==",
       "dev": true,
       "license": "MIT",
       "dependencies": {
@@ -43351,6 +43351,15 @@
         "node": ">=12"
       }
     },
+    "packages/superset-ui-core/node_modules/dompurify": {
+      "version": "3.4.13",
+      "resolved": "https://registry.npmjs.org/dompurify/-/dompurify-3.4.13.tgz",
+      "integrity": "sha512-2vmYIoqjze2d+kakP8S/nS5shfsl587kzwEjcGlTdiksUVgFHnFCsLYDVj/JNqJVOQZGSYBTmuycv0PodwmnMQ==",
+      "license": "(MPL-2.0 OR Apache-2.0)",
+      "optionalDependencies": {
+        "@types/trusted-types": "^2.0.7"
+      }
+    },
     "packages/superset-ui-core/node_modules/react-ace": {
       "version": "14.0.1",
       "resolved": "https://registry.npmjs.org/react-ace/-/react-ace-14.0.1.tgz",

superset-frontend/package.json

@@ -349,7 +349,7 @@
     "swc-loader": "^0.2.7",
     "ts-jest": "^29.4.12",
     "tscw-config": "^1.1.2",
-    "tsx": "^4.23.5",
+    "tsx": "^4.23.7",
     "typescript": "5.4.5",
     "unzipper": "^0.12.5",
     "wait-on": "^9.1.0",

superset-frontend/packages/superset-ui-core/src/components/Select/AsyncSelect.test.tsx

@@ -989,6 +989,52 @@ test('shows all options when filterOption is false', async () => {
   expect(options[0]).toHaveTextContent('Server 0');
 });
 
+test('renders a server-matched option whose label diverges from the search term when filterOption is false (regression for #42041)', async () => {
+  // Mirrors the real permissions-search bug: the remote fetch legitimately
+  // matches the raw, underscore-containing value (e.g. a schema name like
+  // "stg_silver"), but the returned option's displayed label has had
+  // underscores replaced with spaces (see formatPermissionLabel in
+  // features/roles/utils.ts). filterOption defaults to true, which
+  // re-filters already-matched options against that same relabeled text
+  // client-side, so the underscore search term never matches and the
+  // legitimately fetched option gets hidden -- this is why
+  // PermissionsField (features/roles/RoleFormItems.tsx) sets
+  // filterOption={false}: the loader is already the authoritative filter,
+  // and its match doesn't depend on the label used to render the option.
+  const searchData = [{ label: 'stg silver', value: 100 }];
+  const loadOptions = jest.fn(async (search: string) =>
+    // totalCount must exceed the empty initial page here, otherwise
+    // AsyncSelect marks allValuesLoaded and short-circuits every later
+    // fetch, including the search request this test depends on.
+    search === ''
+      ? { data: [], totalCount: 1 }
+      : { data: searchData, totalCount: 1 },
+  );
+
+  render(
+    <AsyncSelect
+      {...defaultProps}
+      options={loadOptions}
+      filterOption={false}
+    />,
+  );
+  await open();
+
+  await type('stg_silver');
+  await waitFor(() =>
+    expect(loadOptions).toHaveBeenCalledWith(
+      'stg_silver',
+      expect.anything(),
+      expect.anything(),
+    ),
+  );
+
+  // The backend legitimately matched and returned this option (asserted
+  // above); it should render in the dropdown despite the search term using
+  // underscores while the label uses spaces.
+  expect(await findSelectOption('stg silver')).toBeInTheDocument();
+});
+
 test('preserves new option entry across search fetch when allowNewOptions is on', async () => {
   const page0Data = Array.from({ length: 10 }, (_, i) => ({
     label: `Option ${i}`,

superset-frontend/packages/superset-ui-core/test/query/getClientErrorObject.test.ts

@@ -336,3 +336,29 @@ test('getErrorText', async () => {
     ),
   ).toEqual('Sorry, an unknown error occurred.');
 });
+
+test('getErrorText for a non-JSON 403 response', async () => {
+  // A 403 originating outside Superset (reverse proxy, WAF, SSO gateway)
+  // carries an HTML or plain-text body instead of the API's JSON
+  // `{"message": "Forbidden"}`, so it must fall back to the generic
+  // status-derived text rather than the permission-denied copy.
+  const proxyForbidden = new Response(
+    '<html><head><title>403 Forbidden</title></head><body>Forbidden</body></html>',
+    {
+      status: 403,
+      statusText: 'Forbidden',
+      headers: { 'Content-Type': 'text/html' },
+    },
+  );
+  expect(await getErrorText(proxyForbidden, 'dashboard')).toEqual(
+    'Sorry, there was an error saving this dashboard: Forbidden',
+  );
+
+  const supersetForbidden = new Response(
+    JSON.stringify({ message: 'Forbidden' }),
+    { status: 403, statusText: 'FORBIDDEN' },
+  );
+  expect(await getErrorText(supersetForbidden, 'dashboard')).toEqual(
+    'You do not have permission to edit this dashboard',
+  );
+});

superset-frontend/playwright/pages/EmbeddedPage.ts

@@ -103,16 +103,40 @@ export class EmbeddedPage {
   /**
    * Wait for dashboard content to render inside the iframe.
    * Looks for the grid-container which indicates charts are loading/loaded.
+   *
+   * Races the grid against the test app's `#error` box so an embed failure
+   * surfaces its message immediately, instead of blindly timing out on the
+   * grid selector and hiding the real reason.
    */
   async waitForDashboardContent(options?: { timeout?: number }): Promise<void> {
-    const frame = this.iframe;
-    await frame
+    const timeout = options?.timeout ?? EMBEDDED.DASHBOARD_RENDER;
+    const grid = this.iframe
       .locator('.grid-container, [data-test="grid-container"]')
-      .first()
-      .waitFor({
-        state: 'visible',
-        timeout: options?.timeout ?? EMBEDDED.DASHBOARD_RENDER,
-      });
+      .first();
+    const errorBox = this.page.locator(EmbeddedPage.SELECTORS.ERROR);
+
+    const ready = grid
+      .waitFor({ state: 'visible', timeout })
+      .then(() => 'ready' as const)
+      .catch(() => 'gridTimeout' as const);
+    const failed = errorBox
+      .waitFor({ state: 'visible', timeout })
+      .then(() => 'error' as const)
+      .catch(() => 'errorTimeout' as const);
+
+    const outcome = await Promise.race([ready, failed]);
+    if (outcome === 'ready') return;
+    if (outcome === 'error') {
+      const message = (await errorBox.textContent())?.trim() || 'unknown error';
+      throw new Error(`Embedded dashboard failed to render: ${message}`);
+    }
+    const status = (
+      await this.page.locator(EmbeddedPage.SELECTORS.STATUS).textContent()
+    )?.trim();
+    throw new Error(
+      `Embedded dashboard did not render within ${timeout}ms ` +
+        `(status: ${status ?? 'unknown'})`,
+    );
   }
 
   /**

superset-frontend/playwright/utils/constants.ts

@@ -101,7 +101,7 @@ export const EMBEDDED = {
   /** Timeout for iframe to appear in the DOM */
   IFRAME_LOAD: 15000, // 15s
   /** Timeout for dashboard content to render inside the iframe */
-  DASHBOARD_RENDER: 30000, // 30s
+  DASHBOARD_RENDER: 60000, // 60s (embedded dashboards are slow to render on cold CI)
   /** Timeout for individual chart cells to finish rendering */
   CHART_RENDER: TIMEOUT.CHART_RENDER,
 } as const;

superset-frontend/plugins/plugin-chart-echarts/src/Timeseries/transformProps.ts

@@ -182,6 +182,50 @@ function getSymbolMarker(symbol: string, color: string) {
   }
 }
 
+/**
+ * Given the fully-built ECharts series (each already carrying its resolved
+ * `stack` id, see `getTimeCompareStackId`), find the largest per-index total
+ * across all series sharing a stack, then return the largest such total
+ * across all stacks. Series without a `stack` id (e.g. annotation layers)
+ * are ignored since they aren't part of any stacked total.
+ */
+function getMaxStackedValueByStack(
+  series: SeriesOption[],
+  isHorizontal: boolean,
+): number {
+  const totalsByStack = new Map<string, number[]>();
+  series.forEach(entry => {
+    const rawStackId = (entry as { stack?: unknown }).stack;
+    const stackId = typeof rawStackId === 'string' ? rawStackId : undefined;
+    if (!stackId || !Array.isArray(entry.data)) return;
+    const totals = totalsByStack.get(stackId) ?? [];
+    (entry.data as unknown[]).forEach((datum, idx) => {
+      let value: unknown = datum;
+      if (Array.isArray(datum)) {
+        value = isHorizontal ? datum[0] : datum[1];
+      } else if (datum && typeof datum === 'object' && 'value' in datum) {
+        const rawValue = (datum as { value: unknown }).value;
+        if (Array.isArray(rawValue)) {
+          value = isHorizontal ? rawValue[0] : rawValue[1];
+        } else {
+          value = rawValue;
+        }
+      }
+      if (typeof value === 'number' && !Number.isNaN(value)) {
+        totals[idx] = (totals[idx] ?? 0) + value;
+      }
+    });
+    totalsByStack.set(stackId, totals);
+  });
+  let max = Number.NEGATIVE_INFINITY;
+  totalsByStack.forEach(totals => {
+    totals.forEach(value => {
+      if (value > max) max = value;
+    });
+  });
+  return max;
+}
+
 export default function transformProps(
   chartProps: EchartsTimeseriesChartProps,
 ): TimeseriesChartTransformedProps {
@@ -875,7 +919,38 @@ export default function transformProps(
   // default to 0-100% range when doing row-level contribution chart
   if ((contributionMode === 'row' || isAreaExpand) && stack) {
     if (yAxisMin === undefined) yAxisMin = 0;
-    if (yAxisMax === undefined) yAxisMax = 1;
+    if (yAxisMax === undefined) {
+      if (contributionMode === 'row') {
+        // Contribution percentages are normalized so each stacked row should
+        // sum to 1, but floating point rounding can push the actual stacked
+        // total fractionally above 1 (e.g. 1.0000000000000002). Hard-capping
+        // the axis max at exactly 1 in that case causes echarts to clip the
+        // topmost stacked segment entirely rather than just rounding the
+        // pixel width, which is most visible in horizontal orientation where
+        // this axis is swapped onto the x-axis. Pad the max up to the actual
+        // stacked total when it exceeds 1 so no segment gets clipped.
+        //
+        // This padding only applies in row-contribution mode: for an Expand
+        // ("100% stacked") chart, `sortedTotalValues` holds the raw,
+        // pre-normalization row totals (e.g. 100), not values near 1, so
+        // padding against them here would stretch the axis out to the raw
+        // total instead of the intended 0-1 range.
+        //
+        // `sortedTotalValues` sums every series value per row regardless of
+        // which ECharts stack it belongs to, but with time_compare each
+        // comparison period is its own independently-normalized stack (see
+        // getTimeCompareStackId), so a chart with N comparison periods would
+        // sum to ~N instead of ~1. Compute the max per stack instead, using
+        // the already-built series (which carry the resolved stack ids).
+        const stackedTotalMax = getMaxStackedValueByStack(series, isHorizontal);
+        yAxisMax =
+          Number.isFinite(stackedTotalMax) && stackedTotalMax > 1
+            ? stackedTotalMax
+            : 1;
+      } else {
+        yAxisMax = 1;
+      }
+    }
   } else if (
     logAxis &&
     yAxisMin === undefined &&

superset-frontend/plugins/plugin-chart-echarts/test/Timeseries/transformProps.test.ts

@@ -23,6 +23,7 @@ import {
   AxisType,
   ChartProps,
   ComparisonType,
+  ContributionType,
   DataRecord,
   EventAnnotationLayer,
   FormulaAnnotationLayer,
@@ -1442,6 +1443,118 @@ test('should not apply axis bounds calculation when seriesType is not Bar for ho
   expect(xAxisRaw.max).toBeUndefined();
 });
 
+test('should not clip small segments when row-contribution percentages float above 1 in horizontal stacked bar charts', () => {
+  // These three shares are individually normalized (each column sums to 1),
+  // but due to floating point rounding their sum can land fractionally
+  // over 1. See https://github.com/apache/superset/issues/30914
+  //
+  // The margin above 1 is chosen large enough (~1e-7) that the sum stays
+  // above 1 no matter which order the underlying series get summed in
+  // (series are sorted by name for stacking, not in the order declared
+  // here), unlike a single-ULP overflow which can round differently
+  // depending on summation order and make this assertion order-dependent.
+  const shareA = 0.42;
+  const shareB = 0.38;
+  const shareC = 0.2000001;
+  expect(shareA + shareB + shareC).toBeGreaterThan(1);
+
+  const queriesData: ChartDataResponseResult[] = [
+    createTestQueryData(
+      createTestData(
+        [{ 'Series A': shareA, 'Series B': shareB, 'Series C': shareC }],
+        { intervalMs: 300000000 },
+      ),
+    ),
+  ];
+
+  const chartProps = createTestChartProps({
+    formData: {
+      ...baseFormDataHorizontalBar,
+      contributionMode: ContributionType.Row,
+      stack: StackControlsValue.Stack,
+    },
+    queriesData,
+  });
+
+  const transformedProps = transformProps(chartProps);
+
+  // In horizontal orientation, axes are swapped, so yAxis becomes xAxis.
+  // The axis max must not be hard-capped at exactly 1, otherwise echarts
+  // clips the topmost stacked segment entirely instead of just rendering
+  // a negligible sub-pixel overflow.
+  const xAxisRaw = transformedProps.echartOptions.xAxis as any;
+  expect(xAxisRaw.max).toBeGreaterThanOrEqual(shareA + shareB + shareC);
+});
+
+test('keeps the 0-1 axis range for Expand (100% stacked) charts instead of padding to the raw row total', () => {
+  // Unlike row-contribution mode, an Expand stack is not pre-normalized in
+  // the query result -- these are raw values (summing to 100, not 1) that
+  // get divided down to a 0-1 range internally. The un-normalized row total
+  // must not be used to pad the axis max, or the chart would only occupy a
+  // sliver of the plot.
+  const queriesData: ChartDataResponseResult[] = [
+    createTestQueryData(
+      createTestData([{ 'Series A': 42, 'Series B': 38, 'Series C': 20 }], {
+        intervalMs: 300000000,
+      }),
+    ),
+  ];
+
+  const chartProps = createTestChartProps({
+    formData: {
+      ...baseFormDataHorizontalBar,
+      stack: StackControlsValue.Expand,
+    },
+    queriesData,
+  });
+
+  const transformedProps = transformProps(chartProps);
+
+  const xAxisRaw = transformedProps.echartOptions.xAxis as any;
+  expect(xAxisRaw.max).toBe(1);
+});
+
+test('computes row-contribution axis padding per stack when time_compare splits a row into multiple normalized stacks', () => {
+  // With time_compare, each comparison period is normalized and stacked
+  // independently (see getTimeCompareStackId), so the current-period
+  // columns sum to ~1 in their own stack and the comparison-period columns
+  // (suffixed with the offset) sum to ~1 in a separate stack. The combined
+  // row total across both stacks is therefore ~2, but the axis max must be
+  // computed per stack, not from that combined total, or a 100% bar would
+  // only occupy about half the plot.
+  const queriesData: ChartDataResponseResult[] = [
+    createTestQueryData(
+      createTestData(
+        [
+          {
+            'Series A': 0.6,
+            'Series B': 0.4,
+            'Series A__1 year ago': 0.55,
+            'Series B__1 year ago': 0.45,
+          },
+        ],
+        { intervalMs: 300000000 },
+      ),
+    ),
+  ];
+
+  const chartProps = createTestChartProps({
+    formData: {
+      ...baseFormDataHorizontalBar,
+      contributionMode: ContributionType.Row,
+      stack: StackControlsValue.Stack,
+      time_compare: ['1 year ago'],
+    },
+    queriesData,
+  });
+
+  const transformedProps = transformProps(chartProps);
+
+  const xAxisRaw = transformedProps.echartOptions.xAxis as any;
+  expect(xAxisRaw.max).toBeGreaterThanOrEqual(1);
+  expect(xAxisRaw.max).toBeLessThan(1.5);
+});
+
 test('legend is visible on tall charts when enabled by the user', () => {
   const chartProps = createTestChartProps({
     height: 400,

superset-frontend/plugins/preset-chart-deckgl/src/layers/common.test.ts

@@ -17,6 +17,7 @@
  * under the License.
  */
 
+import { createElement } from 'react';
 import { PickingInfo } from '@deck.gl/core';
 import { JsonObject, QueryFormData } from '@superset-ui/core';
 import {
@@ -131,6 +132,32 @@ describe('commonLayerProps', () => {
     });
   });
 
+  test('clears a custom tooltip on hover-out instead of trailing the cursor', () => {
+    // Regression test for a custom (Handlebars) deck.gl tooltip that stayed
+    // visible and followed the mouse after leaving a feature.
+    const setTooltip = jest.fn();
+    const customContent = createElement('div', {
+      'data-tooltip-type': 'custom',
+    });
+    const props = commonLayerProps({
+      formData: { ...partialformData } as QueryFormData,
+      setTooltip: setTooltip as any,
+      setTooltipContent: (() => customContent) as any,
+    });
+
+    // Hovering a feature shows the custom tooltip.
+    props.onHover?.({ picked: true, x: 10, y: 20 } as any);
+    expect(setTooltip).toHaveBeenLastCalledWith({
+      content: customContent,
+      x: 10,
+      y: 20,
+    });
+
+    // Moving off the feature must dismiss it, not keep repositioning it.
+    props.onHover?.({ picked: false, x: 30, y: 40 } as any);
+    expect(setTooltip).toHaveBeenLastCalledWith(null);
+  });
+
   test('calls onSelect when table_filter is enabled', () => {
     const formData = {
       ...partialformData,

superset-frontend/plugins/preset-chart-deckgl/src/layers/common.tsx

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-import { ReactNode, isValidElement } from 'react';
+import { ReactNode } from 'react';
 import {
   ascending as d3ascending,
   quantile as d3quantile,
@@ -70,19 +70,16 @@ export function commonLayerProps({
   if (setTooltipContent) {
     let currentTooltipContent: ReactNode = null;
 
-    const isCustomTooltip = (content: ReactNode): boolean =>
-      isValidElement(content) &&
-      content.props?.['data-tooltip-type'] === 'custom';
-
     onHover = (o: JsonObject) => {
       if (o.picked) {
         currentTooltipContent = setTooltipContent(o);
       }
 
-      if (
-        currentTooltipContent &&
-        (o.picked || isCustomTooltip(currentTooltipContent))
-      ) {
+      // Only show the tooltip while a feature is actually hovered. Custom
+      // (Handlebars) tooltips used to stay visible and follow the cursor
+      // after hover-out because their content was kept on screen even when
+      // nothing was picked.
+      if (o.picked && currentTooltipContent) {
         setTooltip({
           content: currentTooltipContent,
           x: o.x,

superset-frontend/src/components/GridTable/index.tsx

@@ -148,7 +148,6 @@ export function GridTable<RecordType extends object>({
       ensureDomOrder: true,
       suppressFieldDotNotation: true,
       headerHeight,
-      rowSelection: 'multiple',
       rowHeight,
     }),
     [rowHeight, headerHeight],

superset-frontend/src/dashboard/actions/dashboardState.test.ts

@@ -372,6 +372,73 @@ describe('dashboardState actions', () => {
         { event: 'dashboard_properties_changed' },
       );
     });
+
+    // The save-error toast mapping lives inline in `onError`, not behind
+    // `getErrorText`, so these exercise the thunk itself. A 403 whose body is
+    // the API's `{"message": "Forbidden"}` shape must surface the
+    // permission-denied copy, while a 403 from outside Superset (reverse proxy,
+    // WAF, SSO gateway) carries a non-JSON body and must fall back to the
+    // generic status-derived toast. See #42239.
+    const findDangerToast = (dispatch: jest.Mock) =>
+      dispatch.mock.calls
+        .map(call => call[0])
+        .find(
+          action =>
+            action?.type === ADD_TOAST &&
+            action.payload.toastType === ToastType.Danger,
+        );
+
+    test('maps a non-JSON 403 save failure to the generic error toast', async () => {
+      const { getState, dispatch } = setup();
+      putStub.mockRestore();
+      putStub = jest.spyOn(SupersetClient, 'put').mockRejectedValue(
+        new Response(
+          '<html><head><title>403 Forbidden</title></head><body>Forbidden</body></html>',
+          {
+            status: 403,
+            statusText: 'Forbidden',
+            headers: { 'Content-Type': 'text/html' },
+          },
+        ),
+      );
+
+      const thunk = saveDashboardRequest(
+        newDashboardData,
+        192,
+        SAVE_TYPE_OVERWRITE,
+      );
+      await thunk(dispatch, getState);
+
+      await waitFor(() =>
+        expect(findDangerToast(dispatch)?.payload.text).toBe(
+          'Sorry, there was an error saving this dashboard: Forbidden',
+        ),
+      );
+    });
+
+    test('maps a Superset JSON 403 save failure to the permission-denied toast', async () => {
+      const { getState, dispatch } = setup();
+      putStub.mockRestore();
+      putStub = jest.spyOn(SupersetClient, 'put').mockRejectedValue(
+        new Response(JSON.stringify({ message: 'Forbidden' }), {
+          status: 403,
+          statusText: 'FORBIDDEN',
+        }),
+      );
+
+      const thunk = saveDashboardRequest(
+        newDashboardData,
+        192,
+        SAVE_TYPE_OVERWRITE,
+      );
+      await thunk(dispatch, getState);
+
+      await waitFor(() =>
+        expect(findDangerToast(dispatch)?.payload.text).toBe(
+          'You do not have permission to edit this dashboard',
+        ),
+      );
+    });
   });
 
   test('fetchCharts returns a Promise that resolves after all refreshes', async () => {

superset-frontend/src/explore/components/controls/AnnotationLayerControl/AnnotationLayer.subdirectory.test.tsx

@@ -54,6 +54,7 @@ const defaultProps = {
   vizType: VizType.Table,
   annotationType: ANNOTATION_TYPES_METADATA.EVENT.value,
   sourceType: 'Table',
+  canReadAnnotation: true,
 };
 
 beforeAll(() => {

superset-frontend/src/explore/components/controls/AnnotationLayerControl/AnnotationLayer.test.tsx

@@ -40,6 +40,7 @@ const defaultProps = {
   value: '',
   vizType: VizType.Table,
   annotationType: ANNOTATION_TYPES_METADATA.FORMULA.value,
+  canReadAnnotation: true,
 };
 
 const nativeLayerApiRoute = 'glob:*/api/v1/annotation_layer/*';
@@ -47,6 +48,11 @@ const chartApiRoute = /\/api\/v1\/chart\/\?q=.+/;
 const chartApiWithIdRoute = /\/api\/v1\/chart\/\w+\?q=.+/;
 
 const chartApiWithIdRouteName = 'chart-with-id';
+const nativeLayerRouteName = 'native-layer';
+
+const nativeLayerResult = {
+  result: [{ name: 'Chart A', id: 'a' }],
+};
 
 const withIdResult = {
   result: {
@@ -107,8 +113,8 @@ beforeAll(() => {
     value => value.value,
   );
 
-  fetchMock.get(nativeLayerApiRoute, {
-    result: [{ name: 'Chart A', id: 'a' }],
+  fetchMock.get(nativeLayerApiRoute, nativeLayerResult, {
+    name: nativeLayerRouteName,
   });
 
   fetchMock.get(chartApiRoute, {
@@ -132,6 +138,12 @@ beforeAll(() => {
   );
 });
 
+// Call history is shared across tests; without this, call-count assertions
+// depend on execution order and fail under `jest --randomize`.
+beforeEach(() => {
+  fetchMock.clearHistory();
+});
+
 const waitForRender = (props?: any) =>
   waitFor(() => render(<AnnotationLayer {...defaultProps} {...props} />));
 
@@ -259,6 +271,137 @@ test('fetches chart on mount if value present', async () => {
   expect(fetchMock.callHistory.calls(chartApiWithIdRoute).length).toBe(1);
 });
 
+test('hides the Superset annotation source without annotation read access', async () => {
+  await waitForRender({
+    annotationType: ANNOTATION_TYPES_METADATA.EVENT.value,
+    canReadAnnotation: false,
+  });
+  userEvent.click(
+    screen.getByRole('combobox', { name: 'Annotation source type' }),
+  );
+  expect(await screen.findByText('Table')).toBeInTheDocument();
+  expect(screen.queryByText('Superset annotation')).not.toBeInTheDocument();
+});
+
+test('keeps formula annotations available without annotation read access', async () => {
+  await waitForRender({ canReadAnnotation: false });
+  expect(screen.getByRole('textbox', { name: 'Formula' })).toBeInTheDocument();
+});
+
+test('keeps a saved native layer intact without annotation read access', async () => {
+  const addAnnotationLayer = jest.fn();
+  await waitForRender({
+    name: 'Test',
+    value: 1,
+    annotationType: ANNOTATION_TYPES_METADATA.EVENT.value,
+    sourceType: 'NATIVE',
+    canReadAnnotation: false,
+    addAnnotationLayer,
+  });
+
+  // The saved source stays selected, and the value select is inert with an
+  // explanation instead of surfacing a Forbidden error.
+  expect(await screen.findByText('Superset annotation')).toBeInTheDocument();
+  expect(
+    screen.getByRole('combobox', { name: 'Annotation layer value' }),
+  ).toBeDisabled();
+  expect(
+    screen.getByText("You don't have permission to view annotation layers."),
+  ).toBeInTheDocument();
+
+  // The saved reference is still valid: re-applying preserves it as is.
+  userEvent.click(screen.getByRole('button', { name: 'Apply' }));
+  expect(addAnnotationLayer).toHaveBeenCalledWith(
+    expect.objectContaining({
+      sourceType: 'NATIVE',
+      value: 1,
+    }),
+  );
+
+  // Neither the by-id fetch nor the listing may fire; both are known 403s.
+  expect(fetchMock.callHistory.calls(nativeLayerApiRoute).length).toBe(0);
+});
+
+test('hydrates the applied native layer name for authorized users', async () => {
+  // The show endpoint returns a single object, unlike the list mock.
+  fetchMock.modifyRoute(nativeLayerRouteName, {
+    response: { result: { id: 1, name: 'My layer' } },
+  });
+
+  try {
+    await waitForRender({
+      name: 'Test',
+      value: 1,
+      annotationType: ANNOTATION_TYPES_METADATA.EVENT.value,
+      sourceType: 'NATIVE',
+    });
+
+    expect(await screen.findByText('My layer')).toBeInTheDocument();
+    expect(fetchMock.callHistory.calls(nativeLayerApiRoute).length).toBe(1);
+  } finally {
+    fetchMock.modifyRoute(nativeLayerRouteName, {
+      response: nativeLayerResult,
+    });
+  }
+});
+
+test('lets a saved native layer switch to a permitted source', async () => {
+  await waitForRender({
+    name: 'Test',
+    value: 1,
+    annotationType: ANNOTATION_TYPES_METADATA.EVENT.value,
+    sourceType: 'NATIVE',
+    canReadAnnotation: false,
+  });
+
+  userEvent.click(
+    screen.getByRole('combobox', { name: 'Annotation source type' }),
+  );
+  userEvent.click(await screen.findByText('Table'));
+
+  // The chart selector takes over, enabled.
+  expect(await screen.findByText('Chart')).toBeInTheDocument();
+  expect(
+    screen.getByRole('combobox', { name: 'Annotation layer value' }),
+  ).toBeEnabled();
+
+  // Reopen the source dropdown: it re-renders from the new options, and the
+  // native option is gone for good.
+  userEvent.click(
+    screen.getByRole('combobox', { name: 'Annotation source type' }),
+  );
+  await waitFor(() =>
+    expect(screen.queryByText('Superset annotation')).not.toBeInTheDocument(),
+  );
+});
+
+test('survives a native annotation layer fetch that fails', async () => {
+  const logError = jest.spyOn(logging, 'error').mockImplementation(() => {});
+  fetchMock.modifyRoute(nativeLayerRouteName, { response: 403 });
+
+  try {
+    await waitForRender({
+      name: 'Test',
+      value: 1,
+      annotationType: ANNOTATION_TYPES_METADATA.EVENT.value,
+      sourceType: 'NATIVE',
+    });
+
+    expect(screen.getByRole('textbox', { name: 'Name' })).toBeInTheDocument();
+    await waitFor(() =>
+      expect(logError).toHaveBeenCalledWith(
+        expect.stringContaining('Failed to load annotation layer 1'),
+        expect.anything(),
+      ),
+    );
+  } finally {
+    fetchMock.modifyRoute(nativeLayerRouteName, {
+      response: nativeLayerResult,
+    });
+    logError.mockRestore();
+  }
+});
+
 test('keeps apply disabled when missing required fields', async () => {
   // With EVENT type and Table source, the component requires selecting a chart
   // and filling in required fields. Without completing these, Apply should be disabled.

superset-frontend/src/explore/components/controls/AnnotationLayerControl/AnnotationLayer.tsx

@@ -30,6 +30,7 @@ import {
   AsyncSelect,
   EmptyState,
   ColorPicker,
+  Typography,
 } from '@superset-ui/core/components';
 import {
   SupersetClient,
@@ -105,6 +106,7 @@ interface AnnotationLayerProps {
   vizType?: string;
   error?: string;
   colorScheme?: string;
+  canReadAnnotation: boolean;
   addAnnotationLayer?: (annotation: Record<string, unknown>) => void;
   removeAnnotationLayer?: () => void;
   close?: () => void;
@@ -166,6 +168,10 @@ const getSliceFormData = (
 const reportChartFailure = (id: string | number) => (error: unknown) =>
   logging.error(`Failed to load annotation source chart ${id}`, error);
 
+const reportAnnotationLayerFailure =
+  (id: string | number) => (error: unknown) =>
+    logging.error(`Failed to load annotation layer ${id}`, error);
+
 const toSliceData = (formData: Record<string, unknown>): SliceData => ({
   data: {
     ...formData,
@@ -232,6 +238,7 @@ function AnnotationLayer({
   vizType,
   error,
   colorScheme = 'd3Category10',
+  canReadAnnotation,
   addAnnotationLayer = () => {},
   removeAnnotationLayer = () => {},
   close = () => {},
@@ -307,17 +314,24 @@ function AnnotationLayer({
           value: key === VizType.Line ? 'line' : key,
           label: chartMetadata?.name || key,
         }));
-      // Prepend native source if applicable
+      // Prepend native source if applicable. Listing native annotation layers
+      // requires can_read on Annotation; without it the option is offered only
+      // while it is the layer's current selection, so a saved native layer
+      // stays intact instead of being silently invalidated.
       const annotationMeta =
         ANNOTATION_TYPES_METADATA[
           annoType as keyof typeof ANNOTATION_TYPES_METADATA
         ];
-      if (annotationMeta && 'supportNativeSource' in annotationMeta) {
+      if (
+        annotationMeta &&
+        'supportNativeSource' in annotationMeta &&
+        (canReadAnnotation || sourceType === ANNOTATION_SOURCE_TYPES.NATIVE)
+      ) {
         sources.unshift(ANNOTATION_SOURCE_TYPES_METADATA.NATIVE);
       }
       return sources;
     },
-    [],
+    [canReadAnnotation, sourceType],
   );
 
   const shouldFetchAppliedAnnotation = useCallback(
@@ -488,14 +502,16 @@ function AnnotationLayer({
     (id: string | number): void => {
       SupersetClient.get({
         endpoint: `/api/v1/annotation_layer/${id}`,
-      }).then(({ json }) => {
-        const { result } = json;
-        const layer = result;
-        setValue({
-          value: layer.id,
-          label: layer.name,
-        });
-      });
+      })
+        .then(({ json }) => {
+          const { result } = json;
+          const layer = result;
+          setValue({
+            value: layer.id,
+            label: layer.name,
+          });
+        })
+        .catch(reportAnnotationLayerFailure(id));
     },
     [],
   );
@@ -503,12 +519,21 @@ function AnnotationLayer({
   const fetchAppliedAnnotation = useCallback(
     (id: string | number): void => {
       if (sourceType === ANNOTATION_SOURCE_TYPES.NATIVE) {
-        fetchAppliedNativeAnnotation(id);
+        // Without can_read on Annotation the request is known to 403; keep the
+        // raw id as the value so the saved layer remains valid and untouched.
+        if (canReadAnnotation) {
+          fetchAppliedNativeAnnotation(id);
+        }
       } else {
         fetchAppliedChart(id);
       }
     },
-    [sourceType, fetchAppliedNativeAnnotation, fetchAppliedChart],
+    [
+      sourceType,
+      canReadAnnotation,
+      fetchAppliedNativeAnnotation,
+      fetchAppliedChart,
+    ],
   );
 
   // componentDidMount - fetch applied annotation if needed
@@ -754,18 +779,34 @@ function AnnotationLayer({
         Example: '2x+5'`);
     }
     if (requiresQuery(sourceType ?? undefined)) {
+      // Listing native annotation layers requires can_read on Annotation.
+      // Keep the select visible but inert so the saved reference can still be
+      // removed, restyled, or switched to a permitted source; the select stays
+      // lazy, so no forbidden request is ever fired.
+      const isBlockedNativeSource =
+        sourceType === ANNOTATION_SOURCE_TYPES.NATIVE && !canReadAnnotation;
       return (
-        <AsyncSelect
-          /* key to force re-render on sourceType change */
-          key={sourceType}
-          ariaLabel={t('Annotation layer value')}
-          name="annotation-layer-value"
-          header={buildChartHeader(label, description, value)}
-          options={fetchOptions}
-          value={value || null}
-          onChange={handleSelectValue}
-          notFoundContent={<NotFoundContent />}
-        />
+        <>
+          <AsyncSelect
+            /* key to force re-render on sourceType change */
+            key={sourceType}
+            ariaLabel={t('Annotation layer value')}
+            name="annotation-layer-value"
+            header={buildChartHeader(label, description, value)}
+            options={fetchOptions}
+            value={value || null}
+            onChange={handleSelectValue}
+            notFoundContent={<NotFoundContent />}
+            disabled={isBlockedNativeSource}
+          />
+          {isBlockedNativeSource && (
+            <div>
+              <Typography.Text type="secondary">
+                {t("You don't have permission to view annotation layers.")}
+              </Typography.Text>
+            </div>
+          )}
+        </>
       );
     }
     if (annotationType === ANNOTATION_TYPES.FORMULA) {
@@ -794,6 +835,7 @@ function AnnotationLayer({
     sourceType,
     annotationType,
     value,
+    canReadAnnotation,
     getSupportedSourceTypes,
     fetchOptions,
     handleSelectValue,

superset-frontend/src/explore/components/controls/AnnotationLayerControl/index.test.tsx

@@ -0,0 +1,52 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { mapStateToProps } from './index';
+
+type State = Parameters<typeof mapStateToProps>[0];
+
+const buildState = (user: unknown): State =>
+  ({
+    charts: {},
+    explore: {
+      controls: {
+        viz_type: { value: 'line' },
+        color_scheme: { value: 'supersetColors' },
+      },
+    },
+    user,
+  }) as unknown as State;
+
+test('grants canReadAnnotation when a role holds can_read on Annotation', () => {
+  const state = buildState({
+    roles: { Gamma: [['can_read', 'Annotation']] },
+  });
+  expect(mapStateToProps(state).canReadAnnotation).toBe(true);
+});
+
+test('denies canReadAnnotation when no role holds the permission', () => {
+  const state = buildState({
+    roles: { Gamma: [['can_read', 'Chart']] },
+  });
+  expect(mapStateToProps(state).canReadAnnotation).toBe(false);
+});
+
+test('denies canReadAnnotation when the user has no roles', () => {
+  expect(mapStateToProps(buildState({})).canReadAnnotation).toBe(false);
+  expect(mapStateToProps(buildState(undefined)).canReadAnnotation).toBe(false);
+});

superset-frontend/src/explore/components/controls/AnnotationLayerControl/index.tsx

@@ -33,6 +33,7 @@ import {
 } from '@superset-ui/core/components';
 import { getChartKey } from 'src/explore/exploreUtils';
 import { runAnnotationQuery } from 'src/components/Chart/chartAction';
+import { findPermission } from 'src/utils/findPermission';
 import CustomListItem from 'src/explore/components/controls/CustomListItem';
 import { ChartState, ExplorePageState } from 'src/explore/types';
 import { AnyAction } from 'redux';
@@ -64,6 +65,7 @@ export interface Props {
   annotationError: Record<string, string>;
   annotationQuery: Record<string, AbortController>;
   vizType: string;
+  canReadAnnotation: boolean;
   validationErrors: JsonObject[];
   name: string;
   actions: {
@@ -85,6 +87,7 @@ function AnnotationLayerControl({
   annotationError = {},
   annotationQuery = {},
   vizType = '',
+  canReadAnnotation,
   validationErrors,
   name,
   actions,
@@ -180,6 +183,7 @@ function AnnotationLayerControl({
             error={error}
             colorScheme={colorScheme}
             vizType={vizType}
+            canReadAnnotation={canReadAnnotation}
             addAnnotationLayer={(newAnnotation: Annotation) =>
               addAnnotationLayer(annotation, newAnnotation)
             }
@@ -195,6 +199,7 @@ function AnnotationLayerControl({
     [
       colorScheme,
       vizType,
+      canReadAnnotation,
       addAnnotationLayer,
       removeAnnotationLayer,
       handleVisibleChange,
@@ -283,10 +288,13 @@ function AnnotationLayerControl({
 
 // Tried to hook this up through stores/control.jsx instead of using redux
 // directly, could not figure out how to get access to the color_scheme
-function mapStateToProps({
+// Exported for tests: the permission wiring below is not covered by tsc
+// (a missing state prop silently falls through to untyped ownProps).
+export function mapStateToProps({
   charts,
   explore,
-}: Pick<ExplorePageState, 'charts' | 'explore'>) {
+  user,
+}: Pick<ExplorePageState, 'charts' | 'explore' | 'user'>) {
   const chartKey = getChartKey(explore);
 
   const defaultChartState: Partial<ChartState> = {
@@ -303,6 +311,9 @@ function mapStateToProps({
     annotationError: chart.annotationError ?? {},
     annotationQuery: chart.annotationQuery ?? {},
     vizType: explore.controls?.viz_type.value,
+    // Mirrors the backend gate on GET /api/v1/annotation_layer/
+    // (class_permission_name "Annotation", get_list -> can_read).
+    canReadAnnotation: findPermission('can_read', 'Annotation', user?.roles),
   };
 }
 

superset-frontend/src/features/roles/RoleFormItems.test.tsx

@@ -16,13 +16,15 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-import { render, screen } from 'spec/helpers/testing-library';
+import { render, screen, waitFor, within } from 'spec/helpers/testing-library';
+import userEvent from '@testing-library/user-event';
 import {
   RoleNameField,
   PermissionsField,
   UsersField,
   GroupsField,
 } from './RoleFormItems';
+import { fetchPermissionOptions } from './utils';
 
 jest.mock('./utils', () => ({
   fetchPermissionOptions: jest.fn(),
@@ -53,6 +55,45 @@ test('PermissionsField renders loading state', () => {
   expect(screen.getByTestId('permissions-select')).toBeInTheDocument();
 });
 
+test('PermissionsField shows a permission matched by its raw name even though the label uses spaces (regression for #42041)', async () => {
+  // fetchPermissionOptions matches the raw, underscore-containing name
+  // server-side; the returned label has already gone through
+  // formatPermissionLabel (underscores replaced with spaces for display).
+  // PermissionsField's normalizing filterOption must match the raw search
+  // term against that space-formatted label, or the option the server
+  // legitimately returned gets hidden by client-side re-filtering.
+  jest
+    .mocked(fetchPermissionOptions)
+    .mockImplementation(async (filterValue: string) =>
+      filterValue === 'stg_silver'
+        ? { data: [{ value: 1, label: 'stg silver' }], totalCount: 1 }
+        : // totalCount must exceed the empty initial page here, otherwise
+          // AsyncSelect marks allValuesLoaded and short-circuits every
+          // later fetch, including the search request this test depends on.
+          { data: [], totalCount: 1 },
+    );
+
+  render(<PermissionsField addDangerToast={addDangerToast} />);
+  const combobox = screen.getByRole('combobox');
+  await waitFor(() => userEvent.click(combobox));
+  await userEvent.clear(combobox);
+  await userEvent.type(combobox, 'stg_silver', { delay: 10 });
+
+  await waitFor(() =>
+    expect(fetchPermissionOptions).toHaveBeenCalledWith(
+      'stg_silver',
+      expect.anything(),
+      expect.anything(),
+      addDangerToast,
+    ),
+  );
+  expect(
+    await within(document.querySelector('.rc-virtual-list')!).findByText(
+      'stg silver',
+    ),
+  ).toBeInTheDocument();
+});
+
 test('UsersField renders label and select', () => {
   render(<UsersField addDangerToast={addDangerToast} loading={false} />);
   expect(screen.getByText('Users')).toBeInTheDocument();

superset-frontend/src/features/roles/RoleFormItems.tsx

@@ -60,6 +60,18 @@ export const PermissionsField = ({
         placeholder={t('Select permissions')}
         options={options}
         loading={loading}
+        // formatPermissionLabel renders the raw permission/view_menu name with
+        // underscores replaced by spaces, so AsyncSelect's default client-side
+        // re-filter never matches a raw-name search term (e.g. "stg_silver")
+        // against the displayed label ("stg silver") and hides the
+        // server-matched option. Normalize both sides so client-side narrowing
+        // still works without hiding valid matches. See #42041.
+        filterOption={(input, option) =>
+          String(option?.label ?? '')
+            .toLowerCase()
+            .replace(/_/g, ' ')
+            .includes(input.toLowerCase().replace(/_/g, ' '))
+        }
         getPopupContainer={trigger => trigger.closest('.ant-modal-container')}
         data-test="permissions-select"
       />

superset/charts/client_processing.py

@@ -25,6 +25,7 @@ In order to do that, we reproduce the post-processing in Python for these chart
 """
 
 import logging
+from functools import partial
 from io import BytesIO, StringIO
 from typing import Any, Optional, TYPE_CHECKING, Union
 
@@ -41,6 +42,12 @@ from superset.utils.core import (
     get_column_names,
     get_metric_names,
 )
+from superset.utils.number_format import (
+    AUTO_CURRENCY,
+    format_number_with_config,
+    resolve_auto_currency,
+    SMART_NUMBER,
+)
 
 if TYPE_CHECKING:
     from superset.connectors.sqla.models import BaseDatasource
@@ -49,6 +56,11 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
+# Default d3 format the Table plugin applies to percent metrics, mirroring
+# ``number-format/NumberFormats.ts::PERCENT_3_POINT`` as used in the frontend's
+# ``transformProps`` formatter selection.
+PERCENT_3_POINT = ",.3%"
+
 
 def get_column_key(label: tuple[str, ...], metrics: list[str]) -> tuple[Any, ...]:
     """
@@ -74,8 +86,9 @@ def pivot_df(  # pylint: disable=too-many-locals, too-many-arguments, too-many-s
     show_rows_total: bool = False,
     show_columns_total: bool = False,
     apply_metrics_on_rows: bool = False,
+    metric_name_aggfunc: Optional[str] = None,
 ) -> pd.DataFrame:
-    metric_name = __("Total (%(aggfunc)s)", aggfunc=aggfunc)
+    metric_name = __("Total (%(aggfunc)s)", aggfunc=metric_name_aggfunc or aggfunc)
 
     if transpose_pivot:
         rows, columns = columns, rows
@@ -198,9 +211,12 @@ def pivot_df(  # pylint: disable=too-many-locals, too-many-arguments, too-many-s
                     )
                     raise
 
-                subtotal = pivot_v2_aggfunc_map[aggfunc](
-                    df.iloc[slice_, :].apply(pd.to_numeric, errors="coerce"), axis=0
-                )
+                subtotal_values = df.iloc[slice_, :]
+                if aggfunc != CURRENCY_CONTEXT_AGGREGATION:
+                    subtotal_values = subtotal_values.apply(
+                        pd.to_numeric, errors="coerce"
+                    )
+                subtotal = pivot_v2_aggfunc_map[aggfunc](subtotal_values, axis=0)
                 depth = groups.nlevels - len(subgroup) - 1
                 total = metric_name if level == 0 else __("Subtotal")
                 subtotal.name = tuple([*subgroup, total, *([""] * depth)])  # noqa: C409
@@ -232,6 +248,41 @@ def list_unique_values(series: pd.Series) -> str:
     return ", ".join({str(v) for v in pd.Series.unique(series)})
 
 
+def union_currency_context(
+    values: Union[pd.Series, pd.DataFrame], axis: int = 0
+) -> Union[tuple[str, ...], pd.Series]:
+    """Union the currency sets contributing to a Pivot Table cell or total."""
+    if isinstance(values, pd.DataFrame):
+        contexts = (
+            [union_currency_context(values[column]) for column in values.columns]
+            if axis == 0
+            else [union_currency_context(values.loc[index]) for index in values.index]
+        )
+        index = values.columns if axis == 0 else values.index
+        return pd.Series(contexts, index=index, dtype=object)
+
+    currencies: dict[str, None] = {}
+    for value in values:
+        if isinstance(value, (list, set, frozenset, tuple)):
+            currencies.update((str(currency), None) for currency in value)
+    return tuple(currencies)
+
+
+CURRENCY_CONTEXT_AGGREGATION = "__currency_context__"
+
+# The frontend's plain Count aggregator is the only Pivot Table aggregator
+# without ``getCurrencies()``. Fraction wrappers inherit that absence, so these
+# modes use the query-wide detected fallback instead of per-cell context.
+PIVOT_AGGREGATIONS_WITHOUT_CURRENCY_CONTEXT = frozenset(
+    {
+        "Count",
+        "Count as Fraction of Total",
+        "Count as Fraction of Rows",
+        "Count as Fraction of Columns",
+    }
+)
+
+
 pivot_v2_aggfunc_map = {
     "Count": pd.Series.count,
     "Count Unique Values": pd.Series.nunique,
@@ -253,53 +304,403 @@ pivot_v2_aggfunc_map = {
     "Count as Fraction of Total": pd.Series.count,
     "Count as Fraction of Rows": pd.Series.count,
     "Count as Fraction of Columns": pd.Series.count,
+    CURRENCY_CONTEXT_AGGREGATION: union_currency_context,
 }
 
 
+def format_column(
+    df: pd.DataFrame,
+    column: Any,
+    d3_format: Optional[str],
+    currency: dict[str, Any],
+    detected_currency: Optional[str] = None,
+    currency_context: Optional[pd.Series] = None,
+    fallback_to_detected: bool = True,
+) -> None:
+    """
+    Format a column in place when a number or currency format is configured.
+
+    ``detected_currency`` represents the query-wide single currency. When a
+    parallel ``currency_context`` series is present, AUTO uses each row/cell's
+    contributing currencies first. Mixed context renders a neutral number;
+    empty context optionally falls back to query-wide detection.
+    """
+    if d3_format or currency.get("symbol"):
+        if currency_context is None:
+            resolved_currency = resolve_auto_currency(currency, detected_currency)
+            df[column] = df[column].apply(
+                partial(format_number_with_config, d3_format, resolved_currency)
+            )
+            return
+
+        contexts = currency_context.reindex(df.index)
+        df[column] = [
+            format_number_with_config(
+                d3_format,
+                resolve_auto_currency(
+                    currency,
+                    detected_currency,
+                    context,
+                    fallback_to_detected,
+                ),
+                value,
+            )
+            for value, context in zip(df[column], contexts, strict=True)
+        ]
+
+
+def get_datasource_column_formats(
+    datasource: Optional[Union["BaseDatasource", "Query"]],
+) -> tuple[dict[str, str | None], dict[str, str]]:
+    """Return saved metric formats and verbose labels from a datasource."""
+    if not datasource:
+        return {}, {}
+
+    datasource_data = datasource.data
+    return (
+        datasource_data.get("column_formats") or {},
+        datasource_data.get("verbose_map") or {},
+    )
+
+
+def get_datasource_currency_formats(
+    datasource: Optional[Union["BaseDatasource", "Query"]],
+) -> tuple[dict[str, dict[str, Any]], dict[str, str]]:
+    """Return saved metric currencies and verbose labels from a datasource.
+
+    The frontend derives ``datasource.currencyFormats`` from each metric's
+    ``currency`` property in ``hydrateExplore.ts``. Report processing receives
+    the raw datasource payload, so it performs the same derivation here.
+    """
+    if not datasource:
+        return {}, {}
+
+    datasource_data = datasource.data
+    stored_currency_formats = datasource_data.get("currency_formats")
+    currency_formats: dict[str, dict[str, Any]] = (
+        {
+            metric: currency
+            for metric, currency in stored_currency_formats.items()
+            if isinstance(metric, str) and isinstance(currency, dict)
+        }
+        if isinstance(stored_currency_formats, dict)
+        else {}
+    )
+    currency_formats.update(
+        {
+            metric["metric_name"]: metric["currency"]
+            for metric in datasource_data.get("metrics") or []
+            if isinstance(metric, dict)
+            and isinstance(metric.get("metric_name"), str)
+            and isinstance(metric.get("currency"), dict)
+            and metric["currency"].get("symbol")
+        }
+    )
+    return currency_formats, datasource_data.get("verbose_map") or {}
+
+
+def get_datasource_currency_column(
+    datasource: Optional[Union["BaseDatasource", "Query"]],
+    df: pd.DataFrame,
+) -> Optional[str]:
+    """Return the currency-code column name as represented in ``df``."""
+    if not datasource:
+        return None
+
+    datasource_data = datasource.data
+    currency_column = datasource_data.get("currency_code_column")
+    if not isinstance(currency_column, str):
+        return None
+    if currency_column in df.columns:
+        return currency_column
+
+    verbose_column = (datasource_data.get("verbose_map") or {}).get(
+        currency_column, currency_column
+    )
+    return verbose_column if verbose_column in df.columns else None
+
+
+def currency_context_value(value: Any) -> tuple[str, ...]:
+    """Convert a truthy row currency value to frontend-compatible context."""
+    try:
+        if value is None or pd.isna(value) or not value:
+            return ()
+    except (TypeError, ValueError):
+        return ()
+    return (str(value),)
+
+
+def build_pivot_currency_context(
+    df: pd.DataFrame,
+    currency_column: str,
+    metrics: list[str],
+    pivot_options: dict[str, Any],
+) -> pd.DataFrame:
+    """
+    Pivot contributing currency sets through the same layout as metric values.
+
+    This mirrors the frontend Pivot Table aggregators' ``currencySet``: every
+    output cell, subtotal, and total carries the union of currencies from its
+    contributing records while numeric aggregation remains unchanged.
+    """
+    currency_source = df.copy()
+    row_context = currency_source[currency_column].map(currency_context_value)
+    for metric in metrics:
+        currency_source[metric] = row_context
+
+    currency_pivot_options = {
+        **pivot_options,
+        "aggfunc": CURRENCY_CONTEXT_AGGREGATION,
+        "metric_name_aggfunc": pivot_options["aggfunc"],
+    }
+    return pivot_df(currency_source, **currency_pivot_options)
+
+
+def get_pivot_currency_format(form_data: dict[str, Any]) -> dict[str, Any]:
+    """Return Pivot Table currency config from transformed or stored form data."""
+    currency_format = form_data.get("currencyFormat") or form_data.get(
+        "currency_format"
+    )
+    return currency_format if isinstance(currency_format, dict) else {}
+
+
+def has_auto_currency_format(
+    form_data: dict[str, Any],
+    datasource: Optional[Union["BaseDatasource", "Query"]] = None,
+) -> bool:
+    """Return whether the Pivot Table has a global or per-metric AUTO format."""
+    currency_formats = [
+        get_pivot_currency_format(form_data),
+        *merge_currency_formats(form_data, datasource).values(),
+    ]
+    return any(
+        isinstance(currency, dict) and currency.get("symbol") == AUTO_CURRENCY
+        for currency in currency_formats
+    )
+
+
+def merge_column_formats(
+    form_data: dict[str, Any],
+    datasource: Optional[Union["BaseDatasource", "Query"]],
+) -> dict[str, str | None]:
+    """Merge saved formats with truthy chart overrides using verbose labels."""
+    saved_formats, verbose_map = get_datasource_column_formats(datasource)
+    column_formats = {
+        verbose_map.get(metric, metric): d3_format
+        for metric, d3_format in saved_formats.items()
+    }
+    column_formats.update(
+        {
+            verbose_map.get(metric, metric): d3_format
+            for metric, d3_format in (form_data.get("columnFormats") or {}).items()
+            if d3_format
+        }
+    )
+    return column_formats
+
+
+def merge_currency_formats(
+    form_data: dict[str, Any],
+    datasource: Optional[Union["BaseDatasource", "Query"]],
+) -> dict[str, dict[str, Any]]:
+    """Merge saved metric currencies with chart overrides by verbose label."""
+    saved_formats, verbose_map = get_datasource_currency_formats(datasource)
+    currency_formats = {
+        verbose_map.get(metric, metric): currency
+        for metric, currency in saved_formats.items()
+    }
+    currency_formats.update(
+        {
+            verbose_map.get(metric, metric): currency
+            for metric, currency in (form_data.get("currencyFormats") or {}).items()
+            if isinstance(currency, dict) and currency.get("symbol")
+        }
+    )
+    return currency_formats
+
+
 def pivot_table_v2(
     df: pd.DataFrame,
     form_data: dict[str, Any],
     datasource: Optional[Union["BaseDatasource", "Query"]] = None,
+    apply_number_format: bool = True,
+    detected_currency: Optional[str] = None,
 ) -> pd.DataFrame:
     """
     Pivot table v2.
     """
     verbose_map = datasource.data["verbose_map"] if datasource else None
+    metrics = get_metric_names(form_data["metrics"], verbose_map)
+    pivot_options: dict[str, Any] = {
+        "rows": get_column_names(form_data.get("groupbyRows"), verbose_map),
+        "columns": get_column_names(form_data.get("groupbyColumns"), verbose_map),
+        "metrics": metrics,
+        "aggfunc": form_data.get("aggregateFunction", "Sum"),
+        "transpose_pivot": bool(form_data.get("transposePivot")),
+        "combine_metrics": bool(form_data.get("combineMetric")),
+        "show_rows_total": bool(form_data.get("rowTotals")),
+        "show_columns_total": bool(form_data.get("colTotals")),
+        "apply_metrics_on_rows": form_data.get("metricsLayout") == "ROWS",
+    }
 
-    return pivot_df(
-        df,
-        rows=get_column_names(form_data.get("groupbyRows"), verbose_map),
-        columns=get_column_names(form_data.get("groupbyColumns"), verbose_map),
-        metrics=get_metric_names(form_data["metrics"], verbose_map),
-        aggfunc=form_data.get("aggregateFunction", "Sum"),
-        transpose_pivot=bool(form_data.get("transposePivot")),
-        combine_metrics=bool(form_data.get("combineMetric")),
-        show_rows_total=bool(form_data.get("rowTotals")),
-        show_columns_total=bool(form_data.get("colTotals")),
-        apply_metrics_on_rows=form_data.get("metricsLayout") == "ROWS",
-    )
+    pivoted = pivot_df(df, **pivot_options)
+    if apply_number_format:
+        currency_context = None
+        if (
+            pivot_options["aggfunc"] not in PIVOT_AGGREGATIONS_WITHOUT_CURRENCY_CONTEXT
+            and has_auto_currency_format(form_data, datasource)
+            and (currency_column := get_datasource_currency_column(datasource, df))
+        ):
+            currency_context = build_pivot_currency_context(
+                df,
+                currency_column,
+                metrics,
+                pivot_options,
+            )
+        return apply_pivot_number_formats(
+            pivoted,
+            form_data,
+            detected_currency,
+            datasource,
+            currency_context,
+        )
+    return pivoted
+
+
+def apply_pivot_number_formats(
+    df: pd.DataFrame,
+    form_data: dict[str, Any],
+    detected_currency: Optional[str] = None,
+    datasource: Optional[Union["BaseDatasource", "Query"]] = None,
+    currency_context: Optional[pd.DataFrame] = None,
+) -> pd.DataFrame:
+    """
+    Apply `valueFormat`/`columnFormats` and currency config to pivot values.
+
+    The metric name is the first column level, or the last when `combineMetric`
+    moves it there; in the ROWS metrics layout it is on the index instead.
+    Per-metric overrides fall back to the global value format.
+    """
+    value_format = form_data.get("valueFormat")
+    column_formats = merge_column_formats(form_data, datasource)
+    currency_format = get_pivot_currency_format(form_data)
+    currency_formats = merge_currency_formats(form_data, datasource)
+    metric_level = -1 if form_data.get("combineMetric") else 0
+    metrics_on_rows = form_data.get("metricsLayout") == "ROWS"
+
+    if metrics_on_rows:
+        df = df.T
+        if currency_context is not None:
+            currency_context = currency_context.T
+
+    for column in df.columns:
+        metric = column[metric_level] if isinstance(column, tuple) else column
+        column_currency_context = (
+            currency_context[column]
+            if currency_context is not None and column in currency_context.columns
+            else None
+        )
+        column_number_format = column_formats.get(metric) or value_format
+        column_currency = currency_formats.get(metric) or currency_format
+        if not column_number_format and not column_currency.get("symbol"):
+            # The frontend Pivot Table formatter falls back to SMART_NUMBER when
+            # neither a value format nor a currency is configured
+            # (``getNumberFormatter``'s default key), so unconfigured metric
+            # cells must not be left raw.
+            column_number_format = SMART_NUMBER
+        format_column(
+            df,
+            column,
+            column_number_format,
+            column_currency,
+            detected_currency,
+            column_currency_context,
+        )
+
+    return df.T if metrics_on_rows else df
 
 
 def table(
     df: pd.DataFrame,
     form_data: dict[str, Any],
-    datasource: Optional[  # pylint: disable=unused-argument
-        Union["BaseDatasource", "Query"]
-    ] = None,
+    datasource: Optional[Union["BaseDatasource", "Query"]] = None,
+    apply_number_format: bool = True,
+    detected_currency: Optional[str] = None,
 ) -> pd.DataFrame:
     """
     Table.
     """
-    # apply `d3NumberFormat` to columns, if present
-    column_config = form_data.get("column_config", {})
-    for column, config in column_config.items():
-        if "d3NumberFormat" in config:
-            format_ = "{:" + config["d3NumberFormat"] + "}"
-            try:
-                df[column] = df[column].apply(format_.format)
-            except Exception:  # pylint: disable=broad-except  # noqa: S110
-                # if we can't format the column for any reason, send as is
-                pass
+    if not apply_number_format:
+        return df
+
+    saved_formats, verbose_map = get_datasource_column_formats(datasource)
+    saved_currency_formats, _ = get_datasource_currency_formats(datasource)
+    currency_column = get_datasource_currency_column(datasource, df)
+    row_currency_context = (
+        df[currency_column].map(currency_context_value) if currency_column else None
+    )
+    column_config = form_data.get("column_config") or {}
+
+    def label_of(name: str) -> str:
+        """Return the column label as it appears in ``df`` (verbose-renamed)."""
+        return name if name in df.columns else verbose_map.get(name, name)
+
+    # Index the per-column overrides by the label present in ``df`` so numeric
+    # and metric columns can be looked up while iterating the frame.
+    number_format_by_label = {
+        label_of(name): fmt for name, fmt in saved_formats.items()
+    }
+    currency_by_label = {
+        label_of(name): currency for name, currency in saved_currency_formats.items()
+    }
+    config_by_label = {label_of(name): config for name, config in column_config.items()}
+    metric_labels = {
+        label_of(name) for name in get_metric_names(form_data.get("metrics"))
+    }
+    # Percent metric columns are emitted with a leading ``%`` and are not
+    # verbose-renamed, so match them by that prefixed label.
+    percent_metric_labels = {
+        f"%{name}"
+        for name in get_metric_names(form_data.get("percent_metrics"))
+        if f"%{name}" in df.columns
+    }
+
+    # Mirror the Table plugin's per-column formatter selection in
+    # ``plugin-chart-table/src/transformProps.ts``: percent metrics default to
+    # PERCENT_3_POINT, every (numeric) metric gets a formatter that defaults to
+    # SMART_NUMBER, and other numeric columns are only formatted when an explicit
+    # format or currency is configured. Dimension and non-numeric columns are
+    # left untouched, matching the browser.
+    for column in df.columns:
+        config = config_by_label.get(column) or {}
+        configured_currency = config.get("currencyFormat") or {}
+        number_format = config.get("d3NumberFormat") or number_format_by_label.get(
+            column
+        )
+        currency = (
+            configured_currency
+            if configured_currency.get("symbol")
+            else currency_by_label.get(column) or {}
+        )
+
+        is_number = pd.api.types.is_numeric_dtype(df[column])
+
+        if column in percent_metric_labels:
+            format_column(df, column, number_format or PERCENT_3_POINT, {})
+        elif (column in metric_labels and is_number) or (
+            is_number and (number_format or currency.get("symbol"))
+        ):
+            if not number_format and not currency.get("symbol"):
+                number_format = SMART_NUMBER
+            format_column(
+                df,
+                column,
+                number_format,
+                currency,
+                detected_currency,
+                row_currency_context,
+                fallback_to_detected=False,
+            )
 
     return df
 
@@ -393,7 +794,14 @@ def apply_client_processing(  # noqa: C901
         if datasource:
             df.rename(columns=datasource.data["verbose_map"], inplace=True)
 
-        processed_df = post_processor(df, form_data, datasource)
+        apply_number_format = query["result_format"] == ChartDataResultFormat.JSON
+        processed_df = post_processor(
+            df,
+            form_data,
+            datasource,
+            apply_number_format,
+            query.get("detected_currency"),
+        )
 
         query["colnames"] = list(processed_df.columns)
         query["indexnames"] = list(processed_df.index)

superset/cli/export_example.py

@@ -155,7 +155,7 @@ def export_example(  # noqa: C901
 
     # Find the dashboard
     if dashboard_id:
-        dashboard = db.session.query(Dashboard).get(dashboard_id)
+        dashboard = db.session.get(Dashboard, dashboard_id)
     elif dashboard_slug:
         dashboard = db.session.query(Dashboard).filter_by(slug=dashboard_slug).first()
     else:

superset/commands/database/sync_permissions.py

@@ -73,6 +73,7 @@ class SyncPermissionsCommand(BaseCommand):
         self.username = username
         self._old_db_connection_name: str | None = old_db_connection_name
         self._db_connection: Database | None = db_connection
+        self._user_id: int | None = None
 
         self.async_mode: bool = app.config["SYNC_DB_PERMISSIONS_IN_ASYNC_MODE"]
 
@@ -99,11 +100,15 @@ class SyncPermissionsCommand(BaseCommand):
         if not self._db_connection:
             raise DatabaseNotFoundError()
 
-        # Need user info to impersonate for OAuth2 connections
-        if not self.username or not security_manager.get_user_by_username(
-            self.username
+        # Need user info to impersonate for OAuth2 connections. The id is
+        # captured here, at validation/enqueue time, so that an async run of
+        # this command binds to whoever held the username right now, rather
+        # than re-resolving the (mutable) username at execution time.
+        if not self.username or not (
+            user := security_manager.get_user_by_username(self.username)
         ):
             raise UserNotFoundInSessionError()
+        self._user_id = user.id
 
         with self.db_connection.get_sqla_engine() as engine:
             try:
@@ -126,7 +131,7 @@ class SyncPermissionsCommand(BaseCommand):
         self.validate()
         if self.async_mode:
             sync_database_permissions_task.delay(
-                self.db_connection_id, self.username, self.old_db_connection_name
+                self.db_connection_id, self._user_id, self.old_db_connection_name
             )
             return
 
@@ -313,14 +318,14 @@ class SyncPermissionsCommand(BaseCommand):
 
 @celery_app.task(name="sync_database_permissions", soft_time_limit=600)
 def sync_database_permissions_task(
-    database_id: int, username: str, old_db_connection_name: str
+    database_id: int, user_id: int, old_db_connection_name: str
 ) -> None:
     """
     Celery task that triggers the SyncPermissionsCommand in async mode.
     """
     with app.test_request_context():
         try:
-            user = security_manager.get_user_by_username(username)
+            user = security_manager.get_user_by_id(user_id)
             if not user:
                 raise UserNotFoundInSessionError()
             g.user = user
@@ -336,7 +341,7 @@ def sync_database_permissions_task(
 
             SyncPermissionsCommand(
                 database_id,
-                username,
+                user.username,
                 old_db_connection_name=old_db_connection_name,
                 db_connection=db_connection,
             ).sync_database_permissions()

superset/commands/database/validate_sql.py

@@ -21,6 +21,7 @@ from typing import Any, Optional
 from flask import current_app as app
 from flask_babel import gettext as __
 
+from superset import security_manager
 from superset.commands.base import BaseCommand
 from superset.commands.database.exceptions import (
     DatabaseNotFoundError,
@@ -69,6 +70,17 @@ class ValidateSQLCommand(BaseCommand):
         schema = self._properties.get("schema")
         template_params = self._properties.get("template_params") or {}
 
+        # Check access before rendering the Jinja template (mirrors the SQL
+        # Lab execute path).
+        security_manager.raise_for_access(
+            database=self._model,
+            sql=sql,
+            catalog=catalog,
+            schema=schema,
+            template_params=template_params,
+            force_dataset_match=True,
+        )
+
         try:
             # Render Jinja templates to handle template syntax before
             # validation. Note: The ENABLE_TEMPLATE_PROCESSING feature flag is

superset/commands/dataset/duplicate.py

@@ -65,7 +65,7 @@ class DuplicateDatasetCommand(CreateMixin, BaseCommand):
         database_id = self._base_model.database_id
         table_name = self._properties["table_name"]
         editors = self._properties["editors"]
-        database = db.session.query(Database).get(database_id)
+        database = db.session.get(Database, database_id)
         if not database:
             raise SupersetErrorException(
                 SupersetError(

superset/commands/dataset/update.py

@@ -123,13 +123,66 @@ class UpdateDatasetCommand(UpdateMixin, BaseCommand):
         # we know we have a valid model
         self._model = cast(SqlaTable, self._model)
         database_id = self._properties.pop("database_id", None)
-        catalog = self._properties.get("catalog")
-        new_db_connection: Database | None = None
-
-        if database_id and database_id != self._model.database.id:
-            if not (new_db_connection := DatasetDAO.get_database_by_id(database_id)):
-                exceptions.append(DatabaseNotFoundValidationError())
+        new_db_connection = self._get_new_database_connection(database_id, exceptions)
         db = new_db_connection or self._model.database
+        database_changed = new_db_connection is not None
+
+        # Detect a caller-supplied change to the source binding, inspected
+        # before the catalog normalization below injects derived values.
+        source_changed = database_changed or any(
+            field in self._properties
+            and self._properties[field] != getattr(self._model, field)
+            for field in ("catalog", "schema", "table_name")
+        )
+
+        catalog, schema, table = self._resolve_catalog_schema_table(db, exceptions)
+
+        # Repointing to a different database connection requires access to
+        # that connection, independent of the caller's editorship of this
+        # dataset -- only persist the change once that's confirmed.
+        if new_db_connection:
+            self._apply_database_repoint(new_db_connection, table, exceptions)
+
+        # Validate uniqueness
+        if not DatasetDAO.validate_update_uniqueness(
+            db,
+            table,
+            self._model_id,
+        ):
+            exceptions.append(DatasetExistsValidationError(table))
+
+        # Repointing a physical dataset (or converting a virtual dataset to a
+        # physical one) runs the same data-access check as the create path.
+        # Skip it when the database connection itself changed: that case is
+        # already covered by the repoint check above, against the same
+        # (db, table) pair.
+        sql = self._properties.get("sql", self._model.sql)
+        if (
+            not new_db_connection
+            and not sql
+            and (source_changed or ("sql" in self._properties and self._model.sql))
+        ):
+            self._validate_table_access(db, table, exceptions)
+
+        self._validate_sql_access(db, catalog, schema, exceptions)
+
+    def _get_new_database_connection(
+        self, database_id: int | None, exceptions: list[ValidationError]
+    ) -> Database | None:
+        # we know we have a valid model
+        self._model = cast(SqlaTable, self._model)
+        if database_id and database_id != self._model.database.id:
+            if new_db_connection := DatasetDAO.get_database_by_id(database_id):
+                return new_db_connection
+            exceptions.append(DatabaseNotFoundValidationError())
+        return None
+
+    def _resolve_catalog_schema_table(
+        self, db: Database, exceptions: list[ValidationError]
+    ) -> tuple[str | None, str | None, Table]:
+        # we know we have a valid model
+        self._model = cast(SqlaTable, self._model)
+        catalog = self._properties.get("catalog")
         default_catalog = db.get_default_catalog()
 
         # If multi-catalog is disabled, and catalog provided is not
@@ -161,29 +214,28 @@ class UpdateDatasetCommand(UpdateMixin, BaseCommand):
             schema,
             catalog,
         )
+        return catalog, schema, table
 
-        # Repointing to a different database connection requires access to
-        # that connection, independent of the caller's editorship of this
-        # dataset -- only persist the change once that's confirmed.
-        if new_db_connection:
-            try:
-                security_manager.raise_for_access(
-                    database=new_db_connection, table=table
-                )
-            except SupersetSecurityException as ex:
-                exceptions.append(DatasetDataAccessIsNotAllowed(ex.error.message))
-            else:
-                self._properties["database"] = new_db_connection
+    def _apply_database_repoint(
+        self,
+        new_db_connection: Database,
+        table: Table,
+        exceptions: list[ValidationError],
+    ) -> None:
+        try:
+            security_manager.raise_for_access(database=new_db_connection, table=table)
+        except SupersetSecurityException as ex:
+            exceptions.append(DatasetDataAccessIsNotAllowed(ex.error.message))
+        else:
+            self._properties["database"] = new_db_connection
 
-        # Validate uniqueness
-        if not DatasetDAO.validate_update_uniqueness(
-            db,
-            table,
-            self._model_id,
-        ):
-            exceptions.append(DatasetExistsValidationError(table))
-
-        self._validate_sql_access(db, catalog, schema, exceptions)
+    def _validate_table_access(
+        self, db: Database, table: Table, exceptions: list[ValidationError]
+    ) -> None:
+        try:
+            security_manager.raise_for_access(database=db, table=table)
+        except SupersetSecurityException as ex:
+            exceptions.append(DatasetDataAccessIsNotAllowed(ex.error.message))
 
     def _validate_sql_access(
         self,
@@ -226,6 +278,9 @@ class UpdateDatasetCommand(UpdateMixin, BaseCommand):
             self._validate_metrics(metrics, exceptions)
             self._validate_expressions(metrics, "metrics", exceptions)
 
+        if predicate := self._properties.get("fetch_values_predicate"):
+            self._validate_fetch_values_predicate(predicate, exceptions)
+
         if folders := self._properties.get("folders"):
             valid_uuids: set[UUID] = set()
             if metrics:
@@ -334,6 +389,34 @@ class UpdateDatasetCommand(UpdateMixin, BaseCommand):
                     )
                 )
 
+    def _validate_fetch_values_predicate(
+        self,
+        predicate: str,
+        exceptions: list[ValidationError],
+    ) -> None:
+        """
+        Validate ``fetch_values_predicate`` with the same parser-based
+        validator used for stored column and metric expressions.
+        """
+        self._model = cast(SqlaTable, self._model)
+        database = self._properties.get("database") or self._model.database
+        catalog = self._properties.get("catalog", self._model.catalog)
+        schema = self._properties.get("schema", self._model.schema)
+        try:
+            validate_stored_expression(database, catalog, schema, predicate)
+        except (SupersetSecurityException, QueryClauseValidationException) as ex:
+            message = (
+                ex.error.message
+                if isinstance(ex, SupersetSecurityException)
+                else ex.message
+            )
+            exceptions.append(
+                ValidationError(
+                    message,
+                    field_name="fetch_values_predicate",
+                )
+            )
+
     @staticmethod
     def _get_duplicates(data: list[dict[str, Any]], key: str) -> list[str]:
         duplicates = [

superset/commands/importers/v1/examples.py

@@ -69,7 +69,7 @@ def transpile_virtual_dataset_sql(config: dict[str, Any], database_id: int) -> N
     if not sql:
         return
 
-    database = db.session.query(Database).get(database_id)
+    database = db.session.get(Database, database_id)
     if not database:
         logger.warning("Database %s not found, skipping SQL transpilation", database_id)
         return

superset/commands/report/alert.py

@@ -39,7 +39,9 @@ from superset.commands.report.exceptions import (
     AlertValidatorConfigError,
     ReportScheduleExecutorNotFoundError,
 )
+from superset.exceptions import SupersetSecurityException
 from superset.reports.models import ReportSchedule, ReportScheduleValidatorType
+from superset.sql.parse import SQLScript
 from superset.tasks.utils import get_executor
 from superset.utils import json
 from superset.utils.core import override_user
@@ -181,6 +183,18 @@ class AlertCommand(BaseCommand):
             "execution_id": self._execution_id,
         }
 
+    def _validate_rendered_sql(self, rendered_sql: str) -> None:
+        """
+        Enforce SQL-level constraints on the rendered alert query: a single
+        statement, and no mutations unless the database allows DML.
+        """
+        database = self._report_schedule.database
+        script = SQLScript(rendered_sql, engine=database.backend)
+        if len(script.statements) != 1:
+            raise AlertQueryError(message=_("Alert query must be a single statement"))
+        if script.has_mutation() and not database.allow_dml:
+            raise AlertQueryError(message=_("Alert query must be read-only"))
+
     @logs_context(context_func=_get_alert_metadata_from_object)
     def _execute_query(self) -> pd.DataFrame:
         """
@@ -196,6 +210,7 @@ class AlertCommand(BaseCommand):
 
         try:
             rendered_sql = sql_template.process_template(self._report_schedule.sql)
+            self._validate_rendered_sql(rendered_sql)
             limited_rendered_sql = self._report_schedule.database.apply_limit_to_sql(
                 rendered_sql, ALERT_SQL_LIMIT
             )
@@ -220,6 +235,18 @@ class AlertCommand(BaseCommand):
                 raise ReportScheduleExecutorNotFoundError(username)
 
             with override_user(user):
+                # Run table-level authorization as the executing user against
+                # the rendered SQL.
+                try:
+                    security_manager.raise_for_access(
+                        database=self._report_schedule.database,
+                        sql=rendered_sql,
+                        force_dataset_match=True,
+                    )
+                except SupersetSecurityException as ex:
+                    raise AlertQueryError(
+                        message=_("Alert query failed the authorization check")
+                    ) from ex
                 start = default_timer()
                 df = self._report_schedule.database.get_df(sql=limited_rendered_sql)
                 stop = default_timer()
@@ -236,6 +263,10 @@ class AlertCommand(BaseCommand):
             # A missing executor user is a configuration problem, not a transient
             # query error; surface the typed error rather than masking it.
             raise
+        except AlertQueryError:
+            # Re-raise the typed validation/authorization errors as-is instead
+            # of masking them behind the generic error below.
+            raise
         except Exception as ex:
             logger.warning("An error occurred when running alert query")
             # The exception message here can reveal to much information to malicious

superset/commands/report/base.py

@@ -15,6 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 import logging
+import re
 from typing import Any, Optional
 
 from croniter import croniter, CroniterBadDateError
@@ -25,6 +26,9 @@ from marshmallow import ValidationError
 from superset import security_manager
 from superset.commands.base import BaseCommand
 from superset.commands.report.exceptions import (
+    AlertQueryDataAccessValidationError,
+    AlertQueryDMLNotAllowedValidationError,
+    AlertQueryMultipleStatementsValidationError,
     ChartNotFoundValidationError,
     ChartNotSavedValidationError,
     DashboardNotFoundValidationError,
@@ -38,16 +42,22 @@ from superset.commands.report.exceptions import (
 from superset.daos.base import BaseDAO
 from superset.daos.chart import ChartDAO
 from superset.daos.dashboard import DashboardDAO
-from superset.exceptions import SupersetSecurityException
+from superset.exceptions import SupersetParseError, SupersetSecurityException
+from superset.models.core import Database
 from superset.reports.models import (
     ReportCreationMethod,
     ReportScheduleType,
 )
 from superset.reports.types import ReportScheduleExtra
+from superset.sql.parse import SQLScript
 from superset.utils import json
 
 logger = logging.getLogger(__name__)
 
+# Matches balanced Jinja blocks so templated alert SQL can be recognized and
+# its static validation deferred to execution time.
+_JINJA_BLOCK_RE = re.compile(r"\{\{.*?\}\}|\{%.*?%\}|\{#.*?#\}", re.DOTALL)
+
 
 class BaseReportScheduleCommand(BaseCommand):
     _properties: dict[str, Any]
@@ -58,6 +68,52 @@ class BaseReportScheduleCommand(BaseCommand):
     def validate(self) -> None:
         pass
 
+    def validate_alert_query(
+        self,
+        database: Database,
+        sql: str,
+        exceptions: list[ValidationError],
+    ) -> None:
+        """
+        Validate alert SQL at save time: it must parse as a single statement,
+        must not mutate state unless the database allows DML, and the saving
+        user must be authorized for the tables it reads. Templated SQL that
+        only parses after rendering is validated at execution time on the
+        rendered query.
+        """
+        contains_jinja = bool(_JINJA_BLOCK_RE.search(sql))
+        try:
+            script = SQLScript(sql, engine=database.backend)
+        except SupersetParseError as ex:
+            if not contains_jinja:
+                exceptions.append(
+                    ValidationError(
+                        _("Invalid SQL: %(error)s", error=ex.error.message),
+                        field_name="sql",
+                    )
+                )
+            return
+        if len(script.statements) != 1:
+            exceptions.append(AlertQueryMultipleStatementsValidationError())
+            return
+        if script.has_mutation() and not database.allow_dml:
+            exceptions.append(AlertQueryDMLNotAllowedValidationError())
+            return
+        try:
+            security_manager.raise_for_access(
+                database=database, sql=sql, force_dataset_match=True
+            )
+        except SupersetSecurityException as ex:
+            exceptions.append(AlertQueryDataAccessValidationError(ex.error.message))
+        except SupersetParseError as ex:
+            if not contains_jinja:
+                exceptions.append(
+                    ValidationError(
+                        _("Invalid SQL: %(error)s", error=ex.error.message),
+                        field_name="sql",
+                    )
+                )
+
     def _check_object_access(
         self,
         object_id: int,

superset/commands/report/create.py

@@ -129,6 +129,8 @@ class CreateReportScheduleCommand(CreateMixin, BaseReportScheduleCommand):
                 database_id = self._properties["database"]
                 if database := DatabaseDAO.find_by_id(database_id):
                     self._properties["database"] = database
+                    if sql := self._properties.get("sql"):
+                        self.validate_alert_query(database, sql, exceptions)
                 else:
                     exceptions.append(DatabaseNotFoundValidationError())
             except KeyError:

superset/commands/report/exceptions.py

@@ -40,6 +40,38 @@ class DatabaseNotFoundValidationError(ValidationError):
         super().__init__(_("Database does not exist"), field_name="database")
 
 
+class AlertQueryMultipleStatementsValidationError(ValidationError):
+    """
+    Marshmallow validation error for alert SQL containing multiple statements
+    """
+
+    def __init__(self) -> None:
+        super().__init__(
+            _("Alert query must be a single statement"),
+            field_name="sql",
+        )
+
+
+class AlertQueryDMLNotAllowedValidationError(ValidationError):
+    """
+    Marshmallow validation error for alert SQL that mutates state on a
+    database that does not allow DML
+    """
+
+    def __init__(self) -> None:
+        super().__init__(_("Alert query must be read-only"), field_name="sql")
+
+
+class AlertQueryDataAccessValidationError(ValidationError):
+    """
+    Marshmallow validation error for alert SQL referencing tables the user
+    is not authorized to query
+    """
+
+    def __init__(self, message: str) -> None:
+        super().__init__(message, field_name="sql")
+
+
 class ReportScheduleDatabaseNotAllowedValidationError(ValidationError):
     """
     Marshmallow validation error for database reference on a Report type schedule

superset/commands/report/execute.py

@@ -57,6 +57,7 @@ from superset.commands.report.exceptions import (
     ReportScheduleXlsxFailedError,
     ReportScheduleXlsxTimeout,
 )
+from superset.commands.report.slack_upgrade import SlackV1UpgradeCoordinator
 from superset.common.chart_data import ChartDataResultFormat, ChartDataResultType
 from superset.daos.report import (
     REPORT_SCHEDULE_ERROR_NOTIFICATION_MARKER,
@@ -81,14 +82,20 @@ from superset.reports.notifications.base import NotificationContent
 from superset.reports.notifications.exceptions import (
     NotificationError,
     NotificationParamException,
-    SlackV1NotificationError,
+)
+from superset.reports.notifications.slack import SlackNotification
+from superset.reports.notifications.slack_transport import (
+    get_slack_send_retry_deadline,
 )
 from superset.subjects.types import SubjectType
 from superset.tasks.utils import get_executor
 from superset.utils import json
-from superset.utils.core import HeaderDataType, override_user, recipients_string_to_list
+from superset.utils.core import HeaderDataType, override_user
 from superset.utils.csv import get_chart_csv_data, get_chart_dataframe
-from superset.utils.decorators import logs_context, transaction
+from superset.utils.decorators import (
+    logs_context,
+    transaction,
+)
 from superset.utils.file import sanitize_title
 from superset.utils.pdf import build_pdf_from_screenshots
 from superset.utils.report_execution import (
@@ -98,7 +105,6 @@ from superset.utils.report_execution import (
     resolve_report_execution_budget_seconds,
 )
 from superset.utils.screenshots import ChartScreenshot, DashboardScreenshot
-from superset.utils.slack import get_channels_with_search, SlackChannelTypes
 from superset.utils.urls import get_url_path
 
 if TYPE_CHECKING:
@@ -279,7 +285,26 @@ class BaseReportState:
         self._start_dttm: datetime = datetime.now(timezone.utc).replace(tzinfo=None)
         self._execution_id = execution_id
         self._report_execution_context = report_execution_context
-        self._filter_warnings: list[str] = []
+        self._execution_warnings: list[str] = []
+        self._slack_v1_upgrade = SlackV1UpgradeCoordinator(
+            report_schedule,
+            execution_id,
+            self._execution_warnings,
+        )
+
+    def _get_slack_retry_deadline(self) -> float:
+        """Return the monotonic deadline for Slack delivery in this execution."""
+        report_deadline = None
+        if self._report_schedule.working_timeout is not None:
+            elapsed = (
+                datetime.now(timezone.utc).replace(tzinfo=None) - self._start_dttm
+            ).total_seconds()
+            remaining = max(
+                float(self._report_schedule.working_timeout) - elapsed,
+                0,
+            )
+            report_deadline = time.monotonic() + remaining
+        return get_slack_send_retry_deadline(report_deadline)
 
     @property
     def _log_context(self) -> str:
@@ -321,13 +346,18 @@ class BaseReportState:
         self,
         state: ReportState,
         error_message: Optional[str] = None,
+        *,
+        include_execution_warnings: bool = True,
     ) -> None:
         """
         Update the report schedule state et al. and reflect the change in the execution
         log.
         """
         self.update_report_schedule(state)
-        self.create_log(error_message)
+        self.create_log(
+            error_message,
+            include_execution_warnings=include_execution_warnings,
+        )
         if state != ReportState.WORKING:
             elapsed, remaining = self._budget_values()
             logger.info(
@@ -359,63 +389,14 @@ class BaseReportState:
         )
 
     def update_report_schedule_slack_v2(self) -> None:
-        """
-        Update the report schedule type and channels for all slack recipients to v2.
-        V2 uses ids instead of names for channels.
-
-        Channel ids for every Slack recipient are resolved first and the
-        recipients are only mutated once all of them resolve. This keeps the
-        upgrade all-or-nothing: a single unresolvable channel can no longer
-        leave the schedule with some recipients already switched to v2 (and
-        persisted by a later error-log commit) while others are untouched.
-        """
-        resolved: list[tuple[ReportRecipients, str]] = []
-        try:
-            for recipient in self._report_schedule.recipients:
-                if recipient.type != ReportRecipientType.SLACK:
-                    continue
-                slack_recipients = json.loads(recipient.recipient_config_json)
-                # V1 method allowed to use leading `#` in the channel name
-                channel_names = (slack_recipients["target"] or "").replace("#", "")
-                # we need to ensure that existing reports can also fetch
-                # ids from private channels
-                channels = get_channels_with_search(
-                    search_string=channel_names,
-                    types=[
-                        SlackChannelTypes.PRIVATE,
-                        SlackChannelTypes.PUBLIC,
-                    ],
-                    exact_match=True,
-                )
-                channels_list = recipients_string_to_list(channel_names)
-                if len(channels_list) != len(channels):
-                    missing_channels = set(channels_list) - {
-                        channel["name"] for channel in channels
-                    }
-                    msg = (
-                        "Could not find the following channels: "
-                        f"{', '.join(missing_channels)}"
-                    )
-                    raise UpdateFailedError(msg)
-                channel_ids = ",".join(channel["id"] for channel in channels)
-                resolved.append((recipient, json.dumps({"target": channel_ids})))
-        except Exception as ex:
-            # No recipient has been mutated yet, so there is no partial upgrade
-            # to revert; surface the failure so the configuration can be fixed
-            # manually.
-            msg = f"Failed to update slack recipients to v2: {str(ex)}"
-            logger.exception(msg)
-            raise UpdateFailedError(msg) from ex
-
-        # Every Slack recipient resolved; apply the upgrade atomically.
-        for recipient, recipient_config_json in resolved:
-            recipient.type = ReportRecipientType.SLACKV2
-            recipient.recipient_config_json = recipient_config_json
+        """Update every Slack v1 recipient atomically to Slack v2."""
+        self._slack_v1_upgrade.update_recipients()
 
     def create_log(
         self,
         error_message: Optional[str] = None,
         *,
+        include_execution_warnings: bool = True,
         log_state: ReportState | None = None,
         reuse_working_log: bool = True,
     ) -> None:
@@ -437,6 +418,15 @@ class BaseReportState:
         """
         from sqlalchemy.orm.exc import StaleDataError
 
+        log_message: Optional[str]
+        if error_message == REPORT_SCHEDULE_ERROR_NOTIFICATION_MARKER:
+            log_message = error_message
+        else:
+            messages = [*self._execution_warnings] if include_execution_warnings else []
+            if error_message:
+                messages.append(error_message)
+            log_message = ";".join(messages) if messages else None
+
         try:
             # Reuse the in-flight WORKING trigger row for this execution, if any,
             # so a single execution surfaces as a single log entry.
@@ -464,7 +454,7 @@ class BaseReportState:
             log.value = self._report_schedule.last_value
             log.value_row_json = self._report_schedule.last_value_row_json
             log.state = effective_state
-            log.error_message = error_message
+            log.error_message = log_message
             db.session.commit()  # pylint: disable=consider-using-transaction
         except StaleDataError as ex:
             # Report schedule was modified or deleted by another process
@@ -586,7 +576,7 @@ class BaseReportState:
                 self._report_schedule.get_native_filters_params()
             )
             if filter_warnings:
-                self._filter_warnings.extend(filter_warnings)
+                self._execution_warnings.extend(filter_warnings)
             if anchor := dashboard_state.get("anchor"):
                 try:
                     anchor_list = json.loads(anchor)
@@ -630,7 +620,7 @@ class BaseReportState:
             self._report_schedule.get_native_filters_params()
         )
         if filter_warnings:
-            self._filter_warnings.extend(filter_warnings)
+            self._execution_warnings.extend(filter_warnings)
         if native_filter_params and native_filter_params != "()":
             # Preserve any urlParams from extra.dashboard (e.g. standalone=true)
             # set via API even when ALERT_REPORT_TABS is off — same merge
@@ -1280,6 +1270,7 @@ class BaseReportState:
                     text=error_text,
                     header_data=header_data,
                     url=url,
+                    slack_retry_deadline=self._get_slack_retry_deadline(),
                     include_cta=include_cta,
                 )
 
@@ -1313,9 +1304,40 @@ class BaseReportState:
             xlsx=xlsx_data,
             embedded_data=embedded_data,
             header_data=header_data,
+            slack_retry_deadline=self._get_slack_retry_deadline(),
             include_cta=include_cta,
         )
 
+    def _send_notification(
+        self,
+        notification_content: NotificationContent,
+        recipient: ReportRecipients,
+    ) -> None:
+        """Send one notification, upgrading Slack v1 recipients when required."""
+        notification = create_notification(recipient, notification_content)
+        if app.config["ALERT_REPORTS_NOTIFICATION_DRY_RUN"]:
+            logger.info(
+                "Would send notification for alert %s, to %s. "
+                "ALERT_REPORTS_NOTIFICATION_DRY_RUN is enabled, "
+                "set it to False to send notifications.",
+                self._report_schedule.name,
+                recipient.recipient_config_json,
+            )
+            return
+
+        if isinstance(notification, SlackNotification):
+            self._slack_v1_upgrade.send(
+                notification,
+                notification_content,
+                create_upgraded_notification=lambda: create_notification(
+                    recipient,
+                    notification_content,
+                ),
+            )
+            return
+
+        notification.send()
+
     def _send(
         self,
         notification_content: NotificationContent,
@@ -1327,54 +1349,34 @@ class BaseReportState:
         :raises: CommandException
         """
         notification_errors: list[SupersetError] = []
+        upgraded_delivery_failed = False
+        self._slack_v1_upgrade.reset()
         report_context = getattr(self, "_report_execution_context", None)
         for recipient in recipients:
-            notification = create_notification(recipient, notification_content)
             try:
-                try:
-                    log_report_delivery_phase(
-                        report_context,
-                        getattr(recipient, "type", None),
-                        "start",
-                        enforce_budget=True,
-                    )
-                    if app.config["ALERT_REPORTS_NOTIFICATION_DRY_RUN"]:
-                        logger.info(
-                            "Would send notification for alert %s, to %s. "
-                            "ALERT_REPORTS_NOTIFICATION_DRY_RUN is enabled, "
-                            "set it to False to send notifications.",
-                            self._report_schedule.name,
-                            recipient.recipient_config_json,
-                        )
-                    else:
-                        notification.send()
-                    log_report_delivery_phase(
-                        report_context,
-                        getattr(recipient, "type", None),
-                        "complete",
-                        enforce_budget=False,
-                    )
-                except SlackV1NotificationError as ex:
-                    # The slack notification should be sent with the v2 api
-                    logger.info(
-                        "Attempting to upgrade the report to Slackv2: %s", str(ex)
-                    )
-                    self.update_report_schedule_slack_v2()
-                    recipient.type = ReportRecipientType.SLACKV2
-                    notification = create_notification(recipient, notification_content)
-                    log_report_delivery_phase(
-                        report_context,
-                        recipient.type,
-                        "retry",
-                        enforce_budget=True,
-                    )
-                    notification.send()
+                log_report_delivery_phase(
+                    report_context,
+                    getattr(recipient, "type", None),
+                    "start",
+                    enforce_budget=True,
+                )
+                self._send_notification(notification_content, recipient)
+                log_report_delivery_phase(
+                    report_context,
+                    getattr(recipient, "type", None),
+                    "complete",
+                    enforce_budget=False,
+                )
             except (
                 UpdateFailedError,
                 NotificationParamException,
                 NotificationError,
                 SupersetException,
             ) as ex:
+                upgraded_delivery_failed = (
+                    upgraded_delivery_failed
+                    or self._slack_v1_upgrade.is_upgraded_recipient(recipient)
+                )
                 # collect errors but keep processing them
                 notification_errors.append(
                     SupersetError(
@@ -1385,6 +1387,12 @@ class BaseReportState:
                         ),
                     )
                 )
+            except Exception:
+                if self._slack_v1_upgrade.is_upgraded_recipient(recipient):
+                    self._slack_v1_upgrade.restore_upgraded_recipients()
+                raise
+        if upgraded_delivery_failed:
+            self._slack_v1_upgrade.restore_upgraded_recipients()
         if notification_errors:
             # log all errors but raise based on the most severe
             for error in notification_errors:
@@ -1735,12 +1743,11 @@ class ReportNotTriggeredErrorState(BaseReportState):
                     )
                     return
             self.send()
-            # Include filter warnings in the log if any were collected
-            warning_message = (
-                ";".join(self._filter_warnings) if self._filter_warnings else None
-            )
             # Clear any retry state from previous failed attempts in this window.
             self._reset_retry_counter()
+            warning_message = (
+                ";".join(self._execution_warnings) if self._execution_warnings else None
+            )
             self.update_report_schedule_and_log(
                 ReportState.SUCCESS, error_message=warning_message
             )
@@ -1810,7 +1817,9 @@ class ReportNotTriggeredErrorState(BaseReportState):
                 finally:
                     try:
                         self.update_report_schedule_and_log(
-                            ReportState.ERROR, error_message=second_error_message
+                            ReportState.ERROR,
+                            error_message=second_error_message,
+                            include_execution_warnings=False,
                         )
                     except ReportScheduleUnexpectedError:
                         # Logging failed again, log it but don't let it hide first_ex
@@ -1985,15 +1994,10 @@ class ReportSuccessState(BaseReportState):
                 raise ex from logging_ex
             raise
 
-        # send() succeeded — clear retry state and log success.
-        # Include filter warnings in the log if any were collected.
-        warning_message = (
-            ";".join(self._filter_warnings) if self._filter_warnings else None
-        )
+        # send() succeeded — clear retry state and log success. Any execution
+        # warnings are incorporated by create_log().
         self._reset_retry_counter()
-        self.update_report_schedule_and_log(
-            ReportState.SUCCESS, error_message=warning_message
-        )
+        self.update_report_schedule_and_log(ReportState.SUCCESS, error_message=None)
 
 
 class ReportScheduleStateMachine:  # pylint: disable=too-few-public-methods

superset/commands/report/slack_upgrade.py

@@ -0,0 +1,225 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import logging
+from collections.abc import Callable
+from uuid import UUID
+
+from flask import current_app as app
+
+from superset.commands.exceptions import UpdateFailedError
+from superset.reports.models import (
+    ReportRecipients,
+    ReportRecipientType,
+    ReportSchedule,
+)
+from superset.reports.notifications.base import BaseNotification, NotificationContent
+from superset.reports.notifications.exceptions import (
+    NotificationParamException,
+    SlackV1NotificationError,
+)
+from superset.reports.notifications.slack import (
+    SLACK_V1_FILE_UPLOAD_MESSAGE,
+    SlackNotification,
+)
+from superset.reports.notifications.slack_channel_resolver import (
+    resolve_slack_channel_ids,
+)
+from superset.utils import json
+from superset.utils.decorators import record_statsd_gauge_failure
+from superset.utils.slack import (
+    NO_SLACK_RECIPIENTS_MESSAGE,
+    parse_slack_recipient_targets,
+    SlackChannelListingClientError,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class SlackV1UpgradeCoordinator:
+    """Coordinate one atomic Slack v1 upgrade and its per-recipient fallbacks."""
+
+    def __init__(
+        self,
+        report_schedule: ReportSchedule,
+        execution_id: UUID,
+        execution_warnings: list[str],
+    ) -> None:
+        self._report_schedule = report_schedule
+        self._execution_id = execution_id
+        self._execution_warnings = execution_warnings
+        self.reset()
+
+    def reset(self) -> None:
+        """Reset execution-scoped upgrade and fallback state."""
+        self._upgrade_error: NotificationParamException | UpdateFailedError | None = (
+            None
+        )
+        self._fallback_recorded = False
+        self._upgraded_recipient_state: list[
+            tuple[ReportRecipients, ReportRecipientType, str]
+        ] = []
+
+    def is_upgraded_recipient(self, recipient: ReportRecipients) -> bool:
+        """Return whether this execution converted the recipient to Slack v2."""
+        return any(
+            upgraded is recipient for upgraded, _, _ in self._upgraded_recipient_state
+        )
+
+    def restore_upgraded_recipients(self) -> None:
+        """Restore recipients when an upgraded Slack delivery does not complete."""
+        for (
+            recipient,
+            recipient_type,
+            recipient_config_json,
+        ) in self._upgraded_recipient_state:
+            recipient.type = recipient_type
+            recipient.recipient_config_json = recipient_config_json
+        self._upgraded_recipient_state = []
+
+    def update_recipients(self) -> None:
+        """Resolve and atomically convert every Slack v1 recipient to Slack v2."""
+        pending: list[tuple[ReportRecipients, list[str]]] = []
+        try:
+            for recipient in self._report_schedule.recipients:
+                if recipient.type != ReportRecipientType.SLACK:
+                    continue
+                try:
+                    slack_recipients = json.loads(recipient.recipient_config_json)
+                except (TypeError, ValueError) as ex:
+                    raise NotificationParamException(
+                        "Invalid Slack recipient configuration"
+                    ) from ex
+                target = (
+                    slack_recipients.get("target")
+                    if isinstance(slack_recipients, dict)
+                    else None
+                )
+                if not isinstance(target, str):
+                    raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
+                channels = parse_slack_recipient_targets(target.replace("#", ""))
+                if not channels:
+                    raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
+                pending.append((recipient, channels))
+
+            all_targets = list(
+                dict.fromkeys(
+                    channel for _, channels in pending for channel in channels
+                )
+            )
+            channel_ids = resolve_slack_channel_ids(all_targets) if all_targets else {}
+            resolved = [
+                (
+                    recipient,
+                    json.dumps(
+                        {
+                            "target": ",".join(
+                                channel_ids[channel] for channel in channels
+                            )
+                        }
+                    ),
+                )
+                for recipient, channels in pending
+            ]
+        except (NotificationParamException, SlackChannelListingClientError) as ex:
+            message = f"Failed to update slack recipients to v2: {ex}"
+            logger.warning(message)
+            raise NotificationParamException(message) from ex
+        except Exception as ex:
+            message = f"Failed to update slack recipients to v2: {ex}"
+            logger.exception(message)
+            raise UpdateFailedError(message) from ex
+
+        self._upgraded_recipient_state = [
+            (recipient, recipient.type, recipient.recipient_config_json)
+            for recipient, _ in resolved
+        ]
+        for recipient, recipient_config_json in resolved:
+            recipient.type = ReportRecipientType.SLACKV2
+            recipient.recipient_config_json = recipient_config_json
+
+    def send_fallback(
+        self,
+        notification: SlackNotification,
+        content: NotificationContent,
+        update_error: NotificationParamException | UpdateFailedError,
+    ) -> None:
+        """Deliver text through Slack v1 and record the first successful fallback."""
+        if content.has_attachments:
+            record_statsd_gauge_failure("reports.slack.send", update_error)
+            message = (
+                f"{SLACK_V1_FILE_UPLOAD_MESSAGE} "
+                f"Slack v2 upgrade failed: {update_error}"
+            )
+            if isinstance(update_error, UpdateFailedError):
+                raise UpdateFailedError(message) from update_error
+            raise NotificationParamException(message) from update_error
+
+        notification.send_legacy_text()
+        if self._fallback_recorded:
+            return
+
+        self._execution_warnings.append(
+            "Slack v2 upgrade unavailable; delivered the text-only report "
+            f"through deprecated Slack v1: {update_error}"
+        )
+        app.config["STATS_LOGGER"].incr("reports.slack.v1_fallback")
+        if isinstance(update_error, UpdateFailedError):
+            app.config["STATS_LOGGER"].incr("reports.slack.v1_fallback.system_error")
+            logger.error(
+                "Slack v2 upgrade failed with a system error; delivered the "
+                "text-only report through Slack v1 for this execution: %s",
+                update_error,
+                extra={
+                    "execution_id": self._execution_id,
+                    "report_schedule_id": self._report_schedule.id,
+                },
+            )
+        else:
+            logger.warning(
+                "Slack v2 upgrade unavailable; delivered the text-only report "
+                "through Slack v1 for this execution: %s",
+                update_error,
+            )
+        self._fallback_recorded = True
+
+    def send(
+        self,
+        notification: SlackNotification,
+        content: NotificationContent,
+        *,
+        create_upgraded_notification: Callable[[], BaseNotification],
+    ) -> None:
+        """Send one Slack v1 recipient, upgrading or falling back when required."""
+        if self._upgrade_error is not None:
+            self.send_fallback(notification, content, self._upgrade_error)
+            return
+
+        try:
+            notification.send()
+        except SlackV1NotificationError as ex:
+            logger.info("Attempting to upgrade the report to Slackv2: %s", ex)
+            try:
+                self.update_recipients()
+            except (
+                NotificationParamException,
+                UpdateFailedError,
+            ) as update_error:
+                self._upgrade_error = update_error
+                self.send_fallback(notification, content, update_error)
+            else:
+                create_upgraded_notification().send()

superset/commands/report/update.py

@@ -149,6 +149,19 @@ class UpdateReportScheduleCommand(UpdateMixin, BaseReportScheduleCommand):
                 exceptions.append(DatabaseNotFoundValidationError())
             self._properties["database"] = database
 
+        # Re-validate the alert SQL whenever the SQL or the target database
+        # changes, using the stored value for whichever half is absent from
+        # the payload.
+        if report_type == ReportScheduleType.ALERT and (
+            "sql" in self._properties or "database" in self._properties
+        ):
+            effective_database = (
+                self._properties.get("database") or self._model.database
+            )
+            effective_sql = self._properties.get("sql", self._model.sql)
+            if effective_database and effective_sql:
+                self.validate_alert_query(effective_database, effective_sql, exceptions)
+
         # validate report frequency
         try:
             self.validate_report_frequency(

superset/commands/sql_lab/estimate.py

@@ -23,8 +23,9 @@ from flask import current_app as app
 from flask_babel import gettext as __
 from jinja2.exceptions import TemplateError
 
-from superset import db, is_feature_enabled, security_manager
+from superset import is_feature_enabled, security_manager
 from superset.commands.base import BaseCommand
+from superset.daos.database import DatabaseDAO
 from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
 from superset.exceptions import (
     SupersetDisallowedSQLFunctionException,
@@ -66,8 +67,10 @@ class QueryEstimationCommand(BaseCommand):
         self._catalog = params.get("catalog")
 
     def validate(self) -> None:
-        self._database = db.session.query(Database).get(self._database_id)
-        if not self._database:
+        # Load the database through the DAO so ``DatabaseFilter`` scopes
+        # visibility the same way it does on the SQL Lab execution path.
+        database = DatabaseDAO.find_by_id(self._database_id)
+        if not database:
             raise SupersetErrorException(
                 SupersetError(
                     message=__("The database could not be found"),
@@ -76,7 +79,17 @@ class QueryEstimationCommand(BaseCommand):
                 ),
                 status=404,
             )
-        security_manager.raise_for_access(database=self._database)
+        self._database = database
+        # Pass the SQL so table-level authorization runs, mirroring the SQL
+        # Lab execution path. Runs before Jinja templating in ``run()``.
+        security_manager.raise_for_access(
+            database=self._database,
+            sql=self._sql,
+            catalog=self._catalog,
+            schema=self._schema or None,
+            template_params=self._template_params,
+            force_dataset_match=True,
+        )
 
     def _apply_sql_security(self, sql: str) -> str:
         """Run the disallowed-function/table, DML and RLS controls against the
@@ -150,6 +163,7 @@ class QueryEstimationCommand(BaseCommand):
 
         sql = self._sql
         if self._template_params:
+            # Access is already checked in validate() before any rendering.
             template_processor = get_template_processor(self._database)
             try:
                 sql = template_processor.process_template(sql, **self._template_params)

superset/commands/streaming_export/base.py

@@ -85,6 +85,14 @@ class BaseStreamingCSVExportCommand(BaseCommand):
         """
         Get the SQL query, database, catalog, and schema for execution.
 
+        The returned SQL is expected to already carry any
+        ``is_split=False`` mutation applied upstream (e.g. by
+        ``get_query_str_extended`` for charts, or SQL Lab's stored
+        ``executed_sql``); ``run()`` applies the complementary
+        ``is_split=True`` mutation exactly once before execution, so the
+        mutator fires once total under either ``MUTATE_AFTER_SPLIT``
+        setting.
+
         Returns:
             Tuple of (sql_query, database_object, catalog, schema)
         """
@@ -223,13 +231,27 @@ class BaseStreamingCSVExportCommand(BaseCommand):
             # Merge database to prevent DetachedInstanceError
             merged_database = session.merge(database)
 
+            # `is_split=True` mirrors the non-streaming download path exactly:
+            # Database.get_df() -> _execute_sql_with_mutation_and_logging()
+            # always calls mutate_sql_based_on_config(..., is_split=True) on
+            # SQL Lab's stored select_sql/executed_sql, and the chart query
+            # path applies its own mutation upstream (in
+            # get_query_str_extended, with is_split=False) before landing
+            # here. Since `is_split` and `MUTATE_AFTER_SPLIT` are compared
+            # for equality, `is_split=True` is the complement of that
+            # upstream chart mutation -- together they mutate the SQL
+            # exactly once for either MUTATE_AFTER_SPLIT setting, instead of
+            # double-mutating when it's False and never mutating when it's
+            # True.
+            mutated_sql = merged_database.mutate_sql_based_on_config(sql, is_split=True)
+
             with merged_database.get_sqla_engine(
                 catalog=catalog, schema=schema
             ) as engine:
                 with engine.connect() as connection:
                     result_proxy = connection.execution_options(
                         stream_results=True
-                    ).execute(text(sql))
+                    ).execute(text(mutated_sql))
 
                     columns = list(result_proxy.keys())
 

superset/common/query_object.py

@@ -440,6 +440,22 @@ class QueryObject:  # pylint: disable=too-many-instance-attributes
         cache_dict: dict[str, Any] = dict(self.to_dict())
         cache_dict.update(extra)
 
+        if "extra_cache_keys" in cache_dict:
+            # Order carries no meaning here (an unordered set of opaque
+            # Jinja url_param()-derived values), but hash_from_dict only
+            # sorts dict keys, not list values, so an unsorted list makes
+            # the cache key depend on Python's per-process hash-randomized
+            # set iteration order (see SqlaTable.get_extra_cache_keys).
+            # Normalize once here so every producer of extra_cache_keys is
+            # safe by construction. Sort on (type name, str value) rather
+            # than a bare str() so values that stringify identically but
+            # differ in type (e.g. 1 and "1") still sort deterministically
+            # instead of falling back to input order.
+            cache_dict["extra_cache_keys"] = sorted(
+                cache_dict["extra_cache_keys"],
+                key=lambda value: (type(value).__name__, str(value)),
+            )
+
         # TODO: the below KVs can all be cleaned up and moved to `to_dict()` at some
         #  predetermined point in time when orgs are aware that the previously
         #  cached results will be invalidated.

superset/config.py

@@ -2582,12 +2582,21 @@ SLACK_CACHE_TIMEOUT = int(timedelta(days=1).total_seconds())
 # For workspaces with 10k+ channels, consider increasing to 10
 SLACK_API_RATE_LIMIT_RETRY_COUNT = 2
 
+# Cooldown (in seconds) after an on-demand Slack channel-cache refresh.
+SLACK_CHANNEL_REFRESH_COOLDOWN_SECONDS = 300
+
 # Timeout (in seconds) for outbound Slack API calls. The Slack SDK defaults to 30s;
 # exposing it here lets operators grant more time for large file uploads (multi-MB
 # CSVs, PDFs, screenshot sets) to congested or rate-limited Slack endpoints without
 # patching code, consistent with the SMTP/CSV/screenshot timeouts.
 SLACK_API_TIMEOUT = 30
 
+# Application retry budget (in seconds) shared by all Slack channels, files, and
+# upload phases in one report execution. Increase this for slow or large-file
+# reports. The effective configured value is floored at one second longer than
+# SLACK_API_TIMEOUT, then clamped to the report's remaining working timeout.
+SLACK_SEND_RETRY_MAX_TIME = 150
+
 # Window size - this will impact the rendering of the data
 WEBDRIVER_WINDOW = {
     "dashboard": (1600, 2000),

superset/connectors/sqla/models.py

@@ -81,6 +81,7 @@ from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
 from superset.exceptions import (
     ColumnNotFoundException,
     DatasetInvalidPermissionEvaluationException,
+    QueryClauseValidationException,
     QueryObjectValidationError,
     SupersetParseError,
     SupersetSecurityException,
@@ -103,6 +104,7 @@ from superset.models.helpers import (
     SoftDeleteMixin,
     SQLA_QUERY_KEYS,
     validate_adhoc_subquery,
+    validate_rendered_expression,
     validate_stored_expression_at_query_time,
 )
 from superset.models.slice import Slice
@@ -1215,6 +1217,14 @@ class TableColumn(AuditMixinNullable, ImportExportMixin, CertificationMixin, Mod
                             msg=msg,
                         )
                     ) from ex
+                if expression != self.expression:
+                    # Re-check the rendered expression before embedding it.
+                    expression = validate_rendered_expression(
+                        expression,
+                        self.database,
+                        self.table.catalog if self.table else None,
+                        self.table.schema if self.table else None,
+                    )
             expression = self._validate_stored_expression(expression)
             col = literal_column(expression, type_=type_)
         else:
@@ -1263,6 +1273,14 @@ class TableColumn(AuditMixinNullable, ImportExportMixin, CertificationMixin, Mod
                             msg=msg,
                         )
                     ) from ex
+                if expression != self.expression:
+                    # Re-check the rendered expression before embedding it.
+                    expression = validate_rendered_expression(
+                        expression,
+                        self.database,
+                        self.table.catalog if self.table else None,
+                        self.table.schema if self.table else None,
+                    )
             expression = self._validate_stored_expression(expression)
             col = literal_column(expression, type_=type_)
         else:
@@ -1368,6 +1386,14 @@ class SqlMetric(AuditMixinNullable, ImportExportMixin, CertificationMixin, Model
                         msg=msg,
                     )
                 ) from ex
+            if expression != self.expression:
+                # Re-check the rendered expression before embedding it.
+                expression = validate_rendered_expression(
+                    expression,
+                    self.table.database,
+                    self.table.catalog,
+                    self.table.schema,
+                )
 
         if expression:
             expression = self._validate_stored_expression(expression)
@@ -1651,7 +1677,18 @@ class SqlaTable(
     def dttm_cols(self) -> list[str]:
         l = [c.column_name for c in self.columns if c.is_dttm]  # noqa: E741
         if self.main_dttm_col and self.main_dttm_col not in l:
-            l.append(self.main_dttm_col)
+            # Only treat ``main_dttm_col`` as a datetime column when the column it
+            # points to is actually temporal. A column whose "Is Temporal" flag was
+            # removed must not keep being reported as a datetime column just because
+            # it is still referenced by ``main_dttm_col`` (#30510). When the column
+            # is not present on the dataset, fall back to the legacy behavior of
+            # trusting ``main_dttm_col``.
+            main_dttm_column: TableColumn | None = next(
+                (c for c in self.columns if c.column_name == self.main_dttm_col),
+                None,
+            )
+            if main_dttm_column is None or main_dttm_column.is_dttm:
+                l.append(self.main_dttm_col)
         return l
 
     @property
@@ -1769,7 +1806,24 @@ class SqlaTable(
                 fetch_values_predicate
             )
         try:
+            # Re-validate the rendered predicate with the same parser policy
+            # as stored column and metric expressions before embedding it.
+            validate_stored_expression(
+                self.database, self.catalog, self.schema, fetch_values_predicate
+            )
             return self.text(fetch_values_predicate)
+        except (SupersetSecurityException, QueryClauseValidationException) as ex:
+            message = (
+                ex.error.message
+                if isinstance(ex, SupersetSecurityException)
+                else ex.message
+            )
+            raise QueryObjectValidationError(
+                _(
+                    "Fetch values predicate failed SQL validation: %(msg)s",
+                    msg=message,
+                )
+            ) from ex
         except (TemplateError, SupersetSyntaxErrorException) as ex:
             msg = getattr(ex, "message", str(ex))
             raise QueryObjectValidationError(
@@ -2474,6 +2528,16 @@ class RowLevelSecurityFilter(Model, AuditMixinNullable):
         Enum(
             *[filter_type.value for filter_type in utils.RowLevelSecurityFilterType],
             name="filter_type_enum",
+            # No migration has ever created a native "filter_type_enum" type in
+            # Postgres - the 2020-09-15 migration that added this column only
+            # ever created a plain VARCHAR. That mismatch was harmless under
+            # SQLAlchemy 1.4, but SQLAlchemy 2.0's postgresql "insertmanyvalues"
+            # feature casts every bound parameter to its column type's DDL name
+            # (`p2::filter_type_enum`) even for a single-row INSERT, which fails
+            # outright since the type doesn't exist. native_enum=False keeps
+            # this a plain VARCHAR (with a CHECK constraint) so the type
+            # actually matches what's really in the database.
+            native_enum=False,
         ),
     )
     group_key = Column(String(255), nullable=True)

superset/daos/dashboard.py

@@ -531,6 +531,15 @@ class DashboardDAO(BaseDAO[Dashboard]):
         dash.params = original_dash.params
         cls.set_dash_metadata(dash, metadata, old_to_new_slice_ids)
         db.session.add(dash)
+        # Flush so the returned dashboard always has a real, persisted
+        # identity (dash.id populated) regardless of what the caller does
+        # next. Without this, whether `dash` ends up with a usable id was an
+        # accident of whatever query the caller happened to run afterward
+        # (autoflush would catch it) - the duplicate_slices=True path leaked
+        # this: it flushes internally per-cloned-slice already, and simple
+        # test/caller code that queries the DB again incidentally
+        # autoflushes too, masking that the plain-copy path never did.
+        db.session.flush()
         return dash
 
     @classmethod

superset/daos/dataset.py

@@ -630,7 +630,7 @@ class DatasetDAO(BaseDAO[SqlaTable]):
         dataset = DatasetDAO.find_by_id(dataset_id)
         if not dataset:
             return None
-        return db.session.query(SqlMetric).get(metric_id)
+        return db.session.get(SqlMetric, metric_id)
 
     @staticmethod
     def get_table_by_name(database_id: int, table_name: str) -> SqlaTable | None:

superset/databases/utils.py

@@ -20,6 +20,7 @@ from __future__ import annotations
 from typing import Any, TYPE_CHECKING
 
 from sqlalchemy.engine.url import make_url, URL
+from sqlalchemy.exc import NoSuchTableError
 
 from superset.commands.database.exceptions import DatabaseInvalidError
 from superset.sql.parse import Table
@@ -72,17 +73,42 @@ def get_table_metadata(database: Any, table: Table) -> TableMetadataResponse:
     :return: Dict table metadata ready for API response
     """
     keys = []
-    columns = database.get_columns(table)
-    primary_key = database.get_pk_constraint(table)
+    table_missing = False
+    try:
+        # get_columns is the table-existence check: SQLAlchemy 2.0's sqlite
+        # dialect raises NoSuchTableError from reflection for a table that
+        # doesn't exist - 1.4's sqlite dialect silently returned empty
+        # results instead, which this API has always relied on to answer
+        # with an empty-but-200 payload for sqlite specifically (other
+        # backends' dialects already raised on missing tables pre-2.0, so
+        # they're unaffected and still surface as the 422 below). Only
+        # sqlite gets the graceful fallback, matching that pre-existing,
+        # dialect-driven difference in behavior between backends. Only this
+        # first call is guarded, so a NoSuchTableError raised later while
+        # reflecting fks/indexes/comments for a table confirmed to exist
+        # still propagates instead of being mistaken for a missing table.
+        columns = database.get_columns(table)
+    except NoSuchTableError:
+        if database.backend != "sqlite":
+            raise
+        table_missing = True
+        columns = []
+    if not table_missing:
+        primary_key = database.get_pk_constraint(table)
+        foreign_keys = get_foreign_keys_metadata(database, table)
+        indexes = get_indexes_metadata(database, table)
+        table_comment = database.get_table_comment(table)
+    else:
+        primary_key = {"constrained_columns": None, "name": None}
+        foreign_keys = []
+        indexes = []
+        table_comment = None
     if primary_key and primary_key.get("constrained_columns"):
         primary_key["column_names"] = primary_key.pop("constrained_columns")
         primary_key["type"] = "pk"
         keys += [primary_key]
-    foreign_keys = get_foreign_keys_metadata(database, table)
-    indexes = get_indexes_metadata(database, table)
     keys += foreign_keys + indexes
     payload_columns: list[TableMetadataColumnsResponse] = []
-    table_comment = database.get_table_comment(table)
     for col in columns:
         dtype = get_col_type(col)
         payload_columns.append(
@@ -102,7 +128,12 @@ def get_table_metadata(database: Any, table: Table) -> TableMetadataResponse:
             show_cols=True if columns else False,
             indent=True,
             cols=columns,
-            latest_partition=True,
+            # A missing table has no partitions to look up, and asking
+            # anyway would just re-trigger the same NoSuchTableError via
+            # select_star()'s own internal database.get_columns() fallback
+            # (it re-fetches columns itself whenever `cols` is empty and
+            # either show_cols or latest_partition is set).
+            latest_partition=not table_missing,
         ),
         "primaryKey": primary_key,
         "foreignKeys": foreign_keys,

superset/db_engine_specs/base.py

@@ -2899,17 +2899,19 @@ class BasicParametersMixin:
         else:
             query.update(cls.encryption_disable_parameters)
 
-        return str(
-            URL.create(
-                f"{cls.engine}+{cls.default_driver}".rstrip("+"),  # type: ignore
-                username=parameters.get("username"),
-                password=parameters.get("password"),
-                host=parameters["host"],
-                port=parameters["port"],
-                database=parameters["database"],
-                query=query,
-            )
-        )
+        # SQLAlchemy 2.0 made URL.__str__() hide the password by default
+        # (it rendered in full under 1.4); render_as_string(hide_password=
+        # False) is required here since this URI is stored/used to actually
+        # connect, not just displayed.
+        return URL.create(
+            f"{cls.engine}+{cls.default_driver}".rstrip("+"),  # type: ignore
+            username=parameters.get("username"),
+            password=parameters.get("password"),
+            host=parameters["host"],
+            port=parameters["port"],
+            database=parameters["database"],
+            query=query,
+        ).render_as_string(hide_password=False)
 
     @classmethod
     def get_parameters_from_uri(  # pylint: disable=unused-argument

superset/db_engine_specs/clickhouse.py

@@ -487,17 +487,19 @@ class ClickHouseConnectEngineSpec(BasicParametersMixin, ClickHouseEngineSpec):
         if not url_params.get("database"):
             url_params["database"] = "__default__"
 
-        return str(
-            URL.create(
-                f"{cls.engine}+{cls.default_driver}",
-                username=url_params.get("username"),
-                password=url_params.get("password"),
-                host=url_params.get("host"),
-                port=url_params.get("port"),
-                database=url_params.get("database"),
-                query=url_params.get("query"),
-            )
-        )
+        # SQLAlchemy 2.0 made URL.__str__() hide the password by default
+        # (it rendered in full under 1.4); render_as_string(hide_password=
+        # False) is required here since this URI is stored/used to actually
+        # connect, not just displayed.
+        return URL.create(
+            f"{cls.engine}+{cls.default_driver}",
+            username=url_params.get("username"),
+            password=url_params.get("password"),
+            host=url_params.get("host"),
+            port=url_params.get("port"),
+            database=url_params.get("database"),
+            query=url_params.get("query"),
+        ).render_as_string(hide_password=False)
 
     @classmethod
     def get_parameters_from_uri(

superset/db_engine_specs/couchbase.py

@@ -176,7 +176,11 @@ class CouchbaseEngineSpec(BasicParametersMixin, BaseEngineSpec):
                 query=query_params,
             )
         print(uri)
-        return str(uri)
+        # SQLAlchemy 2.0 made URL.__str__() hide the password by default
+        # (it rendered in full under 1.4); render_as_string(hide_password=
+        # False) is required here since this URI is stored/used to actually
+        # connect, not just displayed.
+        return uri.render_as_string(hide_password=False)
 
     @classmethod
     def get_parameters_from_uri(

superset/db_engine_specs/databend.py

@@ -282,17 +282,19 @@ class DatabendEngineSpec(BasicParametersMixin, DatabendBaseEngineSpec):
             else cls.encryption_disable_parameters
         )
 
-        return str(
-            URL.create(
-                cls.engine,
-                username=parameters.get("username"),
-                password=parameters.get("password"),
-                host=parameters.get("host"),
-                port=parameters.get("port"),
-                database=parameters.get("database") or "__default__",
-                query=query,
-            )
-        )
+        # SQLAlchemy 2.0 made URL.__str__() hide the password by default
+        # (it rendered in full under 1.4); render_as_string(hide_password=
+        # False) is required here since this URI is stored/used to actually
+        # connect, not just displayed.
+        return URL.create(
+            cls.engine,
+            username=parameters.get("username"),
+            password=parameters.get("password"),
+            host=parameters.get("host"),
+            port=parameters.get("port"),
+            database=parameters.get("database") or "__default__",
+            query=query,
+        ).render_as_string(hide_password=False)
 
     @classmethod
     def _encryption_from_tls_parameters(

superset/db_engine_specs/databricks.py

@@ -664,17 +664,19 @@ class DatabricksNativeEngineSpec(DatabricksDynamicBaseEngineSpec):
                 )
             query.update(cls.encryption_parameters)
 
-        return str(
-            URL.create(
-                f"{cls.engine}+{cls.default_driver}".rstrip("+"),
-                username="token",
-                password=parameters.get("access_token"),
-                host=parameters["host"],
-                port=parameters["port"],
-                database=parameters["database"],
-                query=query,
-            )
-        )
+        # SQLAlchemy 2.0 made URL.__str__() hide the password by default
+        # (it rendered in full under 1.4); render_as_string(hide_password=
+        # False) is required here since this URI is stored/used to actually
+        # connect, not just displayed.
+        return URL.create(
+            f"{cls.engine}+{cls.default_driver}".rstrip("+"),
+            username="token",
+            password=parameters.get("access_token"),
+            host=parameters["host"],
+            port=parameters["port"],
+            database=parameters["database"],
+            query=query,
+        ).render_as_string(hide_password=False)
 
     @classmethod
     def get_parameters_from_uri(  # type: ignore
@@ -891,16 +893,18 @@ class DatabricksPythonConnectorEngineSpec(DatabricksDynamicBaseEngineSpec):
         if parameters.get("encryption"):
             query.update(cls.encryption_parameters)
 
-        return str(
-            URL.create(
-                cls.engine,
-                username="token",
-                password=parameters.get("access_token"),
-                host=parameters["host"],
-                port=parameters["port"],
-                query=query,
-            )
-        )
+        # SQLAlchemy 2.0 made URL.__str__() hide the password by default
+        # (it rendered in full under 1.4); render_as_string(hide_password=
+        # False) is required here since this URI is stored/used to actually
+        # connect, not just displayed.
+        return URL.create(
+            cls.engine,
+            username="token",
+            password=parameters.get("access_token"),
+            host=parameters["host"],
+            port=parameters["port"],
+            query=query,
+        ).render_as_string(hide_password=False)
 
     @classmethod
     def get_parameters_from_uri(  # type: ignore

superset/db_engine_specs/duckdb.py

@@ -125,7 +125,11 @@ class DuckDBParametersMixin:
         ):
             return MotherDuckEngineSpec.build_sqlalchemy_uri(parameters)
 
-        return str(URL(drivername=cls.engine, database=database, query=query))
+        # SQLAlchemy 2.0 made URL a strict NamedTuple - the raw URL(...)
+        # constructor now requires username/password/host/port to be passed
+        # explicitly (they used to default to None). URL.create() keeps
+        # those optional, matching the pre-2.0 URL(...) behavior used here.
+        return str(URL.create(drivername=cls.engine, database=database, query=query))
 
     @classmethod
     def get_parameters_from_uri(  # pylint: disable=unused-argument
@@ -443,8 +447,14 @@ class MotherDuckEngineSpec(DuckDBEngineSpec):
                 f"Need MotherDuck token to connect to database '{database}'."
             )
 
+        # SQLAlchemy 2.0 made URL a strict NamedTuple - the raw URL(...)
+        # constructor now requires username/password/host/port to be passed
+        # explicitly (they used to default to None). URL.create() keeps
+        # those optional, matching the pre-2.0 URL(...) behavior used here.
         return str(
-            URL(drivername=DuckDBEngineSpec.engine, database=database, query=query)
+            URL.create(
+                drivername=DuckDBEngineSpec.engine, database=database, query=query
+            )
         )
 
     @classmethod

superset/db_engine_specs/postgres.py

@@ -641,12 +641,11 @@ class PostgresEngineSpec(BasicParametersMixin, PostgresBaseEngineSpec):
         """
         Return the default schema for a given query.
 
-        This method simply uses the parent method after checking that there are no
-        malicious path setting in the query.
+        This method simply uses the parent method after checking that the query
+        cannot rebind the schema used to resolve unqualified table names.
         """
         script = process_jinja_sql(query.sql, database, template_params).script
-        settings = script.get_settings()
-        if "search_path" in settings:
+        if script.changes_default_schema():
             raise SupersetSecurityException(
                 SupersetError(
                     error_type=SupersetErrorType.QUERY_SECURITY_ACCESS_ERROR,

superset/db_engine_specs/snowflake.py

@@ -356,19 +356,21 @@ class SnowflakeEngineSpec(PostgresBaseEngineSpec):
             dict[str, Any]
         ] = None,
     ) -> str:
-        return str(
-            URL.create(
-                "snowflake",
-                username=parameters.get("username"),
-                password=parameters.get("password"),
-                host=parameters.get("account"),
-                database=parameters.get("database"),
-                query={
-                    "role": parameters.get("role"),
-                    "warehouse": parameters.get("warehouse"),
-                },
-            )
-        )
+        # SQLAlchemy 2.0 made URL.__str__() hide the password by default
+        # (it rendered in full under 1.4); render_as_string(hide_password=
+        # False) is required here since this URI is stored/used to actually
+        # connect, not just displayed.
+        return URL.create(
+            "snowflake",
+            username=parameters.get("username"),
+            password=parameters.get("password"),
+            host=parameters.get("account"),
+            database=parameters.get("database"),
+            query={
+                "role": parameters.get("role"),
+                "warehouse": parameters.get("warehouse"),
+            },
+        ).render_as_string(hide_password=False)
 
     @classmethod
     def get_parameters_from_uri(

superset/extensions/__init__.py

@@ -154,7 +154,19 @@ async_query_manager: AsyncQueryManager = LocalProxy(
 cache_manager = CacheManager()
 celery_app = celery.Celery()
 csrf = CSRFProtect()
-db = get_sqla_class()()
+
+# Flask-SQLAlchemy 3.x scopes db.session by the identity of the current Flask
+# app-context object (id(app_ctx)) rather than by thread/greenlet identity like
+# 2.x did. Superset's codebase (and its test fixtures) widely assumes a single
+# shared session per thread across nested `app.app_context()` blocks, often
+# relying on that implicit sharing instead of an explicit commit. Restoring the
+# 2.x scopefunc here keeps that assumption valid under FSA 3.x.
+try:
+    from greenlet import getcurrent as _session_scopefunc
+except ImportError:
+    from threading import get_ident as _session_scopefunc
+
+db = get_sqla_class()(session_options={"scopefunc": _session_scopefunc})
 
 # make_versioned() MUST be called immediately after db is constructed and before
 # any versioned model class is defined.  Continuum patches the SQLAlchemy

superset/initialization/__init__.py

@@ -26,7 +26,16 @@ from typing import Any, Callable, TYPE_CHECKING
 import wtforms_json
 from colorama import Fore, Style
 from deprecation import deprecated
-from flask import abort, current_app, Flask, redirect, request, session, url_for
+from flask import (
+    abort,
+    current_app,
+    Flask,
+    has_app_context,
+    redirect,
+    request,
+    session,
+    url_for,
+)
 from flask_appbuilder import expose, IndexView
 from flask_appbuilder.api import safe
 from flask_appbuilder.utils.base import get_safe_redirect
@@ -147,8 +156,16 @@ class SupersetAppInitializer:  # pylint: disable=too-many-public-methods
             # pylint: disable=too-few-public-methods
             abstract = True
 
-            # Grab each call into the task and set up an app context
+            # Grab each call into the task and set up an app context, unless
+            # one is already active on this thread (e.g. Celery eager mode
+            # invoked from within an existing request/test context) - Flask-
+            # SQLAlchemy 3.x scopes db.session by the active app context's
+            # object identity rather than by thread, so pushing a redundant
+            # nested context here would silently hand the task a second,
+            # blind session unable to see the caller's uncommitted work.
             def __call__(self, *args: Any, **kwargs: Any) -> Any:
+                if has_app_context():
+                    return task_base.__call__(self, *args, **kwargs)
                 with superset_app.app_context():
                     return task_base.__call__(self, *args, **kwargs)
 

superset/jinja_context.py

@@ -915,6 +915,18 @@ class BaseTemplateProcessor:
         """
         return self._context.copy()
 
+    def get_template_context(self, **kwargs: Any) -> dict[str, Any]:
+        """
+        Build the validated context used to render a template.
+
+        Split out from ``process_template`` so that validation paths which
+        render a pre-parsed template (``superset.sql.parse.process_jinja_sql``)
+        use exactly the same context as execution, keeping the validated SQL
+        identical to the executed SQL.
+        """
+        kwargs.update(self._context)
+        return validate_template_context(self.engine, kwargs)
+
     def process_template(self, sql: str, **kwargs: Any) -> str:
         """Processes a sql template
 
@@ -984,8 +996,7 @@ class BaseTemplateProcessor:
 
             raise SupersetTemplateException(message) from ex
 
-        kwargs.update(self._context)
-        context = validate_template_context(self.engine, kwargs)
+        context = self.get_template_context(**kwargs)
 
         try:
             return template.render(context)
@@ -1133,27 +1144,21 @@ class HiveTemplateProcessor(PrestoTemplateProcessor):
 class SparkTemplateProcessor(HiveTemplateProcessor):
     engine = "spark"
 
-    def process_template(self, sql: str, **kwargs: Any) -> str:
-        template = self.env.from_string(sql)
-        kwargs.update(self._context)
-
+    def get_template_context(self, **kwargs: Any) -> dict[str, Any]:
+        context = super().get_template_context(**kwargs)
         # Backwards compatibility if migrating from Hive.
-        context = validate_template_context(self.engine, kwargs)
         context["hive"] = context["spark"]
-        return template.render(context)
+        return context
 
 
 class TrinoTemplateProcessor(PrestoTemplateProcessor):
     engine = "trino"
 
-    def process_template(self, sql: str, **kwargs: Any) -> str:
-        template = self.env.from_string(sql)
-        kwargs.update(self._context)
-
+    def get_template_context(self, **kwargs: Any) -> dict[str, Any]:
+        context = super().get_template_context(**kwargs)
         # Backwards compatibility if migrating from Presto.
-        context = validate_template_context(self.engine, kwargs)
         context["presto"] = context["trino"]
-        return template.render(context)
+        return context
 
 
 DEFAULT_PROCESSORS = {

superset/key_value/shared_entries.py

@@ -35,6 +35,7 @@ RESOURCE = KeyValueResource.APP
 CODEC = JsonKeyValueCodec()
 
 
+@transaction()
 def get_shared_value(key: SharedKey) -> Optional[Any]:
     """
     Get a shared value by key, with configurable fallback for backward compatibility.

superset/mcp_service/chart/preview_utils.py

@@ -69,7 +69,7 @@ def generate_preview_from_form_data(
         from superset.connectors.sqla.models import SqlaTable
         from superset.extensions import db
 
-        dataset = db.session.query(SqlaTable).get(dataset_id)
+        dataset = db.session.get(SqlaTable, dataset_id)
         if not dataset:
             return ChartError(
                 error=f"Dataset {dataset_id} not found", error_type="DatasetNotFound"

superset/mcp_service/chart/schemas.py

@@ -36,6 +36,7 @@ from pydantic import (
     field_validator,
     model_serializer,
     model_validator,
+    StrictBool,
     ValidationError,
 )
 from typing_extensions import Self
@@ -2202,6 +2203,19 @@ class ListChartsRequest(
 ):
     """Request schema for list_charts with clear, unambiguous types."""
 
+    certified: Annotated[
+        StrictBool | None,
+        Field(
+            default=None,
+            description=(
+                "Filter by governance certification status. Use true to return "
+                "only certified charts (preferred when selecting governed "
+                "assets), false to return only uncertified charts, or omit to "
+                "return both (default)."
+            ),
+        ),
+    ]
+
     deleted_state: Annotated[
         Literal["include", "only"] | None,
         Field(

superset/mcp_service/chart/tool/list_charts.py

@@ -80,7 +80,9 @@ async def list_charts(
     """List charts with filtering and search.
 
     Returns chart metadata including id, name, viz_type, URL, and last
-    modified time.
+    modified time. Set ``request.certified`` to true to return only governed
+    charts; false returns only uncertified charts, while omitting it preserves
+    the unfiltered behavior.
 
     **IMPORTANT**: All parameters must be wrapped in a ``request`` object.
     Do NOT pass ``search``, ``page``, ``page_size``, etc. as top-level
@@ -124,7 +126,7 @@ async def list_charts(
         )
     )
 
-    from superset.charts.filters import ChartDeletedStateFilter
+    from superset.charts.filters import ChartCertifiedFilter, ChartDeletedStateFilter
     from superset.daos.chart import ChartDAO
     from superset.mcp_service.common.schema_discovery import (
         CHART_SORTABLE_COLUMNS,
@@ -179,6 +181,13 @@ async def list_charts(
 
     try:
         with event_logger.log_context(action="mcp.list_charts.query"):
+            custom_filters = None
+            if request.certified is not None:
+                custom_filters = {
+                    "certified": tool.build_bound_filter(
+                        ChartCertifiedFilter, request.certified
+                    )
+                }
             result = tool.run_tool(
                 filters=request.filters,
                 search=request.search,
@@ -190,6 +199,7 @@ async def list_charts(
                 created_by_me=request.created_by_me,
                 edited_by_me=request.edited_by_me,
                 deleted_state=request.deleted_state,
+                custom_filters=custom_filters,
             )
         count = len(result.charts) if hasattr(result, "charts") else 0
         total_pages = getattr(result, "total_pages", None)

superset/mcp_service/dataset/schemas.py

@@ -32,6 +32,7 @@ from pydantic import (
     field_validator,
     model_serializer,
     model_validator,
+    StrictBool,
 )
 
 from superset.daos.base import ColumnOperator, ColumnOperatorEnum
@@ -246,6 +247,19 @@ class ListDatasetsRequest(
     test_list_datasets_with_string_filters.
     """
 
+    certified: Annotated[
+        StrictBool | None,
+        Field(
+            default=None,
+            description=(
+                "Filter by governance certification status. Use true to return "
+                "only certified datasets (preferred when selecting governed "
+                "semantic-layer assets), false to return only uncertified "
+                "datasets, or omit to return both (default)."
+            ),
+        ),
+    ]
+
     @field_validator("filters", mode="before")
     @classmethod
     def parse_filters(cls, v: Any) -> Any:

superset/mcp_service/dataset/tool/list_datasets.py

@@ -94,7 +94,9 @@ async def list_datasets(
     """List datasets with filtering and search.
 
     Returns dataset metadata including table name, schema, and last modified
-    time.
+    time. Set ``request.certified`` to true to return only governed,
+    semantic-layer datasets; false returns only uncertified datasets, while
+    omitting it preserves the unfiltered behavior.
 
     **IMPORTANT**: All parameters must be wrapped in a ``request`` object.
     Do NOT pass ``search``, ``page``, ``page_size``, etc. as top-level
@@ -160,6 +162,7 @@ async def list_datasets(
 
     try:
         from superset.daos.dataset import DatasetDAO
+        from superset.datasets.filters import DatasetCertifiedFilter
         from superset.mcp_service.common.schema_discovery import (
             DATASET_SORTABLE_COLUMNS,
             get_all_column_names,
@@ -191,6 +194,13 @@ async def list_datasets(
         )
 
         with event_logger.log_context(action="mcp.list_datasets.query"):
+            custom_filters = None
+            if request.certified is not None:
+                custom_filters = {
+                    "certified": tool.build_bound_filter(
+                        DatasetCertifiedFilter, request.certified
+                    )
+                }
             result = tool.run_tool(
                 filters=request.filters,
                 search=request.search,
@@ -201,6 +211,7 @@ async def list_datasets(
                 page_size=request.page_size,
                 created_by_me=request.created_by_me,
                 edited_by_me=request.edited_by_me,
+                custom_filters=custom_filters,
             )
 
         await ctx.info(

superset/mcp_service/mcp_core.py

@@ -115,28 +115,30 @@ class BaseCore(ABC):
         self.logger.warning(message)
 
 
-class DeletedStateBoundFilter:
-    """Adapt a FAB deleted-state filter for ``BaseDAO.list`` custom_filters.
+class BoundFilter:
+    """Bind a caller value to a FAB filter used by ``BaseDAO.list``.
 
     ``BaseDAO.list`` invokes custom filters as ``apply(query, None)``, but the
-    ``BaseDeletedStateFilter`` subclasses interpret ``None`` as "live rows
-    only". Binding the value at construction lets the DAO-side invocation
-    reach the FAB filter with the caller's actual ``include``/``only`` choice.
-
-    ``model`` re-exposes the FAB filter's SoftDeleteMixin model class so the
-    caller can scope the session visibility bypass without re-consulting the
-    (Optional) filter-class attribute.
+    request value is already available to MCP callers. Binding it at
+    construction preserves that value for the DAO-side invocation.
     """
 
-    def __init__(self, inner: Any, value: str, model: type) -> None:
+    def __init__(self, inner: Any, value: Any) -> None:
         self._inner = inner
         self._value = value
-        self.model = model
 
     def apply(self, query: Any, value: Any) -> Any:
         return self._inner.apply(query, self._value)
 
 
+class DeletedStateBoundFilter(BoundFilter):
+    """Bound deleted-state filter carrying its visibility-bypass model."""
+
+    def __init__(self, inner: Any, value: str, model: type) -> None:
+        super().__init__(inner, value)
+        self.model = model
+
+
 class ModelListCore(BaseCore, Generic[L]):
     """
     Generic tool for listing model objects with filtering, search, pagination, and
@@ -340,6 +342,11 @@ class ModelListCore(BaseCore, Generic[L]):
         inner = self._deleted_state_filter("id", datamodel)
         return DeletedStateBoundFilter(inner, normalized, model)
 
+    def build_bound_filter(self, filter_class: type, value: Any) -> BoundFilter:
+        """Bind an MCP value to a FAB filter for this core's DAO model."""
+        datamodel = SQLAInterface(self.dao_class.model_cls, db.session)
+        return BoundFilter(filter_class("id", datamodel), value)
+
     def run_tool(
         self,
         filters: Any | None = None,
@@ -352,6 +359,7 @@ class ModelListCore(BaseCore, Generic[L]):
         created_by_me: bool = False,
         edited_by_me: bool = False,
         deleted_state: str | None = None,
+        custom_filters: Dict[str, Any] | None = None,
     ) -> L:
         # Clamp page_size to MAX_PAGE_SIZE as defense-in-depth
         page_size = min(page_size, MAX_PAGE_SIZE)
@@ -401,7 +409,9 @@ class ModelListCore(BaseCore, Generic[L]):
             "search": search,
             "columns_to_load": columns_to_load,
         }
+        dao_custom_filters = dict(custom_filters or {})
         if deleted_state_bound is not None:
+            dao_custom_filters["deleted_state"] = deleted_state_bound
             # The soft-delete ORM listener appends ``deleted_at IS NULL`` at
             # execution time, so the session-scoped bypass must span both
             # executions inside DAO.list (count + fetch). The context manager
@@ -410,11 +420,13 @@ class ModelListCore(BaseCore, Generic[L]):
             # which unhidden rows the caller may actually see.
             with skip_visibility_filter(db.session, deleted_state_bound.model):
                 items, total_count = self._call_dao_list(
-                    custom_filters={"deleted_state": deleted_state_bound},
+                    custom_filters=dao_custom_filters,
                     **dao_kwargs,
                 )
         else:
-            items, total_count = self._call_dao_list(**dao_kwargs)
+            items, total_count = self._call_dao_list(
+                custom_filters=dao_custom_filters or None, **dao_kwargs
+            )
         # Serialize items
         item_objs = []
         for item in items:

superset/migrations/versions/2018-07-26_11-10_c82ee8a39623_add_implicit_tags.py

@@ -30,7 +30,7 @@ from datetime import datetime  # noqa: E402
 
 from alembic import op  # noqa: E402
 from sqlalchemy import Column, DateTime, Enum, ForeignKey, Integer, String  # noqa: E402
-from sqlalchemy.orm import declarative_base, declared_attr  # noqa: E402
+from sqlalchemy.orm import declarative_base, declared_attr, Mapped  # noqa: E402
 
 from superset.tags.models import ObjectType, TagType  # noqa: E402
 from superset.utils.core import get_user_id  # noqa: E402
@@ -55,7 +55,7 @@ class AuditMixinNullable:
     )
 
     @declared_attr
-    def created_by_fk(self) -> Column:
+    def created_by_fk(self) -> Mapped[int | None]:
         return Column(
             Integer,
             ForeignKey("ab_user.id"),
@@ -64,7 +64,7 @@ class AuditMixinNullable:
         )
 
     @declared_attr
-    def changed_by_fk(self) -> Column:
+    def changed_by_fk(self) -> Mapped[int | None]:
         return Column(
             Integer,
             ForeignKey("ab_user.id"),

superset/migrations/versions/2020-01-08_01-17_e96dbf2cfef0_datasource_cluster_fk.py

@@ -44,7 +44,7 @@ def upgrade():
         batch_op.add_column(sa.Column("cluster_id", sa.Integer()))
 
     # Update cluster_id values
-    metadata = sa.MetaData(bind=bind)
+    metadata = sa.MetaData()
     datasources = sa.Table("datasources", metadata, autoload_with=bind)
     clusters = sa.Table("clusters", metadata, autoload_with=bind)
 
@@ -86,7 +86,7 @@ def downgrade():
         batch_op.add_column(sa.Column("cluster_name", sa.String(250)))
 
     # Update cluster_name values
-    metadata = sa.MetaData(bind=bind)
+    metadata = sa.MetaData()
     datasources = sa.Table("datasources", metadata, autoload_with=bind)
     clusters = sa.Table("clusters", metadata, autoload_with=bind)
 

superset/migrations/versions/2020-04-24_10-46_e557699a813e_add_tables_relation_to_row_level_.py

@@ -35,7 +35,7 @@ down_revision = "743a117f0d98"
 
 def upgrade():
     bind = op.get_bind()
-    metadata = sa.MetaData(bind=bind)
+    metadata = sa.MetaData()
     insp = sa.engine.reflection.Inspector.from_engine(bind)
 
     rls_filter_tables = create_table(
@@ -53,7 +53,7 @@ def upgrade():
 
     for row in bind.execute(filter_ids):
         move_table_id = rls_filter_tables.insert().values(
-            rls_filter_id=row["id"], table_id=row["table_id"]
+            rls_filter_id=row._mapping["id"], table_id=row._mapping["table_id"]
         )
         bind.execute(move_table_id)
 
@@ -68,7 +68,7 @@ def upgrade():
 
 def downgrade():
     bind = op.get_bind()
-    metadata = sa.MetaData(bind=bind)
+    metadata = sa.MetaData()
 
     op.add_column(
         "row_level_security_filters",

superset/migrations/versions/2020-09-15_18-22_e5ef6828ac4e_add_rls_filter_type_and_grouping_key.py

@@ -43,7 +43,7 @@ def upgrade():
         )
 
     bind = op.get_bind()
-    metadata = sa.MetaData(bind=bind)
+    metadata = sa.MetaData()
     filters = sa.Table("row_level_security_filters", metadata, autoload_with=bind)
     statement = filters.update().values(
         filter_type=utils.RowLevelSecurityFilterType.REGULAR.value

superset/migrations/versions/2020-09-24_12-04_3fbbc6e8d654_fix_data_access_permissions_for_virtual_.py

@@ -168,7 +168,7 @@ def upgrade():  # noqa: C901
         match_ds_id = re.match(r"\[None\]\.\[.*\]\(id:(\d+)\)", faulty_view_menu.name)
         if match_ds_id:
             dataset_id = int(match_ds_id.group(1))
-            dataset = session.query(SqlaTable).get(dataset_id)
+            dataset = session.get(SqlaTable, dataset_id)
             if dataset:
                 try:
                     new_view_menu = dataset.get_perm()

superset/migrations/versions/2020-09-28_17-57_b56500de1855_add_uuid_column_to_import_mixin.py

@@ -140,10 +140,11 @@ def upgrade():
             batch_op.create_unique_constraint(f"uq_{table_name}_uuid", ["uuid"])
 
     # add UUID to Dashboard.position_json
+    slices_model = models["slices"]
     slice_uuid_map = {
         slc.id: slc.uuid
-        for slc in session.query(models["slices"])
-        .options(load_only("id", "uuid"))
+        for slc in session.query(slices_model)
+        .options(load_only(slices_model.id, slices_model.uuid))
         .all()
     }
     update_dashboards(session, slice_uuid_map)

superset/migrations/versions/2021-02-04_09-34_070c043f2fdb_add_granularity_to_charts_where_missing.py

@@ -92,7 +92,7 @@ def upgrade():
             if "granularity" in params or "granularity_sqla" in params:
                 continue
 
-            table = session.query(SqlaTable).get(slc.datasource_id)
+            table = session.get(SqlaTable, slc.datasource_id)
             if not table:
                 continue
 

superset/migrations/versions/2021-02-18_09-13_c501b7c653a3_add_missing_uuid_column.py

@@ -87,10 +87,11 @@ def upgrade():
 
     # add UUID to Dashboard.position_json; this function is idempotent
     # so we can call it for all objects
+    slices_model = models["slices"]
     slice_uuid_map = {
         slc.id: slc.uuid
-        for slc in session.query(models["slices"])
-        .options(load_only("id", "uuid"))
+        for slc in session.query(slices_model)
+        .options(load_only(slices_model.id, slices_model.uuid))
         .all()
     }
     update_dashboards(session, slice_uuid_map)

superset/migrations/versions/2022-04-01_14-38_a9422eeaae74_new_dataset_models_take_2.py

@@ -35,6 +35,7 @@ from sqlalchemy.orm import (
     backref,
     declarative_base,
     declared_attr,
+    Mapped,
     relationship,
     Session,
 )
@@ -191,7 +192,7 @@ class SqlaTable(AuxiliaryColumnsMixin, Base):
     id = sa.Column(sa.Integer, primary_key=True)
     extra = sa.Column(sa.Text)
     database_id = sa.Column(sa.Integer, sa.ForeignKey("dbs.id"), nullable=False)
-    database: Database = relationship(
+    database: Mapped[Database] = relationship(
         "Database",
         backref=backref("tables", cascade="all, delete-orphan"),
         foreign_keys=[database_id],
@@ -274,7 +275,7 @@ class NewTable(AuxiliaryColumnsMixin, Base):
     name = sa.Column(sa.Text)
     external_url = sa.Column(sa.Text, nullable=True)
     extra_json = sa.Column(MediumText(), default="{}")
-    database: Database = relationship(
+    database: Mapped[Database] = relationship(
         "Database",
         backref=backref("new_tables", cascade="all, delete-orphan"),
         foreign_keys=[database_id],

superset/models/core.py

@@ -32,7 +32,6 @@ from datetime import datetime
 from functools import lru_cache
 from inspect import signature
 from typing import Any, Callable, cast, Optional, TYPE_CHECKING
-from urllib.parse import quote
 
 import numpy
 import pandas as pd
@@ -506,7 +505,14 @@ class Database(CoreDatabase, AuditMixinNullable, ImportExportMixin):  # pylint:
             # do not over-write the password with the password mask
             self.password = conn.password
         conn = conn.set(password=PASSWORD_MASK if conn.password else None)
-        self.sqlalchemy_uri = str(conn)  # hides the password
+        # Store the literal PASSWORD_MASK sentinel (not the real secret -
+        # that already went to self.password above), so later code that
+        # compares conn.password against PASSWORD_MASK to detect an
+        # unchanged password keeps working. SQLAlchemy 2.0 changed
+        # str(URL) to substitute its own "***" for any password rather
+        # than rendering the value verbatim (str(conn) under 1.4), so
+        # render_as_string(hide_password=False) is required here.
+        self.sqlalchemy_uri = conn.render_as_string(hide_password=False)
 
     def get_effective_user(self, object_url: URL) -> str | None:
         """
@@ -709,7 +715,14 @@ class Database(CoreDatabase, AuditMixinNullable, ImportExportMixin):  # pylint:
         if cacheable and self.id is not None:
             cache_key = (
                 self.id,
-                str(sqlalchemy_url),
+                # SQLAlchemy 2.0 changed str(URL) to always substitute
+                # "***" for the password rather than rendering the real
+                # value (str(url) under 1.4). Using it here would make
+                # the cache key blind to password rotation - the module
+                # comment above depends on the key changing when the
+                # password does, so render_as_string(hide_password=False)
+                # is required to preserve that behavior.
+                sqlalchemy_url.render_as_string(hide_password=False),
                 repr(sorted(engine_kwargs.items())),
             )
             with _ENGINE_CACHE_LOCK:
@@ -1379,13 +1392,15 @@ class Database(CoreDatabase, AuditMixinNullable, ImportExportMixin):  # pylint:
         else:
             raw_password = self.password
 
-        # Encode the password such that special characters
-        # are preserved when rendering to string and reparsing the URL.
-        if raw_password is not None:
-            encoded_password = quote(raw_password, safe="")
-            conn = conn.set(password=encoded_password)
-        else:
-            conn = conn.set(password=None)
+        # URL.render_as_string() percent-encodes the password itself, so pass
+        # the raw, un-encoded password straight through. Pre-encoding it here
+        # (as this used to do) is safe under SQLAlchemy 1.4, whose
+        # render_as_string() treats URL.password as literal, but SQLAlchemy
+        # 2.0 always encodes on render - double-encoding a pre-escaped
+        # password (e.g. "%" -> "%25") turns it into "%2525" on write, which
+        # then decodes back to the wrong value ("%25" instead of "%") on the
+        # next reparse.
+        conn = conn.set(password=raw_password)
 
         # render_as_string preserves the URL encoding of special
         # characters in passwords

superset/models/helpers.py

@@ -153,7 +153,7 @@ from superset.utils.date_parser import (
     TimeDeltaAmbiguousError,
 )
 from superset.utils.dates import datetime_to_epoch
-from superset.utils.rls import apply_rls
+from superset.utils.rls import apply_rls, get_predicates_for_table
 
 
 class ValidationResultDict(TypedDict):
@@ -329,6 +329,62 @@ def validate_stored_expression_at_query_time(
     return expression
 
 
+def validate_rendered_expression(
+    expression: str,
+    database: Database,
+    catalog: str | None,
+    schema: str | None,
+) -> str:
+    """
+    Apply the stored-expression validation policy to a rendered expression.
+
+    Query-time counterpart to ``validate_stored_expression``: it runs on the
+    already-rendered expression that is embedded via ``literal_column`` and
+    applies the same policy, failing closed on unparseable results.
+
+    :param expression: the rendered expression
+    :returns: the expression to embed, possibly rewritten with RLS predicates
+    :raises QueryObjectValidationError: on multi-statement, set-operation,
+        disallowed sub-query, or sanitization failures -- matching the
+        ``QueryObjectValidationError`` contract callers already expect from
+        ``validate_stored_expression_at_query_time``, rather than letting a
+        raw ``SupersetSecurityException`` escape uncaught.
+    """
+    engine = database.backend
+    wrapped = f"SELECT {expression}"
+
+    try:
+        parsed = SQLStatement(wrapped, engine)
+    except SupersetParseError as ex:
+        raise QueryObjectValidationError(
+            _("Custom SQL fields cannot be parsed as a single SQL statement.")
+        ) from ex
+
+    if parsed.is_set_operation():
+        raise QueryObjectValidationError(
+            _("Custom SQL fields cannot contain set operations.")
+        )
+
+    try:
+        wrapped = validate_adhoc_subquery(
+            wrapped, database, catalog, schema or "", engine
+        )
+    except SupersetSecurityException as ex:
+        raise QueryObjectValidationError(ex.message) from ex
+    try:
+        wrapped = sanitize_clause(wrapped, engine)
+    except QueryClauseValidationException as ex:
+        raise QueryObjectValidationError(ex.message) from ex
+
+    prefix, expression = re.split(
+        r"SELECT\s+",
+        wrapped,
+        maxsplit=1,
+        flags=re.IGNORECASE,
+    )
+    return expression.strip()
+
+
 def json_to_dict(json_str: str) -> dict[Any, Any]:
     if json_str:
         val = re.sub(",[ \t\r\n]+}", "}", json_str)
@@ -3083,9 +3139,40 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
                 if rls_applied:
                     from_sql = parsed_script.format()
 
-            except Exception as ex:
-                # Log the error but don't fail - RLS application is best-effort
-                logger.warning("Failed to apply RLS to virtual dataset SQL: %s", ex)
+            except Exception as ex:  # pylint: disable=broad-except
+                # RLS injection failures fail closed: only continue when it is
+                # positively confirmed that no RLS predicates apply to the
+                # referenced tables; any other outcome aborts the query.
+                try:
+                    rls_required = any(
+                        get_predicates_for_table(
+                            table.qualify(
+                                catalog=self.catalog,
+                                schema=self.schema or default_schema or "",
+                            ),
+                            self.database,
+                            self.database.get_default_catalog(),
+                            exclude_dataset_id=self_id,
+                        )
+                        for statement in parsed_script.statements
+                        for table in statement.tables
+                    )
+                except Exception:  # pylint: disable=broad-except
+                    rls_required = True
+                if rls_required:
+                    raise QueryObjectValidationError(
+                        _(
+                            "Row-level security could not be applied to the "
+                            "virtual dataset query, so it cannot be run "
+                            "securely: %(msg)s",
+                            msg=str(ex),
+                        )
+                    ) from ex
+                logger.warning(
+                    "RLS application to virtual dataset SQL failed, but no "
+                    "predicates apply to its tables; continuing: %s",
+                    ex,
+                )
 
         cte = self.db_engine_spec.get_cte_query(from_sql)
         from_clause = (
@@ -3782,6 +3869,11 @@ class ExploreMixin:  # pylint: disable=too-many-public-methods
         if expression := tbl_column.expression:
             if template_processor:
                 expression = template_processor.process_template(expression)
+                if expression != tbl_column.expression:
+                    # Re-check the rendered expression before embedding it.
+                    expression = validate_rendered_expression(
+                        expression, self.database, self.catalog, self.schema
+                    )
             expression = self._validate_stored_expression(expression)
             col = literal_column(expression, type_=type_)
         else:

superset/models/sql_types/presto_sql_types.py

@@ -16,7 +16,7 @@
 # under the License.
 
 # pylint: disable=abstract-method
-from typing import Any, Optional
+from typing import Any, Callable, Optional
 
 from sqlalchemy.engine.interfaces import Dialect
 from sqlalchemy.sql.sqltypes import DATE, Integer, TIMESTAMP
@@ -113,6 +113,23 @@ class TimeStamp(TypeDecorator):
         """
         return f"TIMESTAMP '{value}'"
 
+    def literal_processor(self, dialect: Dialect) -> Callable[[str], str]:
+        """
+        Used when compiling with literal_binds=True (e.g. where_latest_partition).
+
+        TypeDecorator's standard composition model (overriding
+        process_literal_param/process_bind_param) expects those hooks to
+        return a plain value that the impl TIMESTAMP type's own processor
+        then converts to SQL text - but process_bind_param here already
+        returns the final literal text ("TIMESTAMP '...'"). Composing that
+        through TIMESTAMP's literal_processor breaks under SQLAlchemy 2.0
+        (it expects a real datetime and raises CompileError on the string).
+        Overriding literal_processor directly - against the base class's own
+        advice - is the only way to keep this class's "process_bind_param
+        already produces final SQL text" design working.
+        """
+        return lambda value: self.process_bind_param(value, dialect)
+
 
 class Date(TypeDecorator):
     """
@@ -129,3 +146,12 @@ class Date(TypeDecorator):
         as Presto does not support automatic casting.
         """
         return f"DATE '{value}'"
+
+    def literal_processor(self, dialect: Dialect) -> Callable[[str], str]:
+        """
+        Used when compiling with literal_binds=True (e.g. where_latest_partition).
+
+        See TimeStamp.literal_processor above for why this override is
+        needed under SQLAlchemy 2.0.
+        """
+        return lambda value: self.process_bind_param(value, dialect)

superset/reports/notifications/base.py

@@ -35,11 +35,17 @@ class NotificationContent:
     description: Optional[str] = ""
     url: Optional[str] = None  # url to chart/dashboard for this screenshot
     embedded_data: Optional[pd.DataFrame] = None
+    slack_retry_deadline: Optional[float] = None
     # Populated only when this is a per-retry or final-failure notification
     retry_attempt: Optional[int] = None
     retry_max_attempts: Optional[int] = None
     include_cta: bool = True  # include the call-to-action link back to Superset
 
+    @property
+    def has_attachments(self) -> bool:
+        """Return whether the notification contains any file attachment."""
+        return bool(self.csv or self.xlsx or self.pdf or self.screenshots)
+
 
 class BaseNotification:  # pylint: disable=too-few-public-methods
     """

superset/reports/notifications/exceptions.py

@@ -24,6 +24,10 @@ class NotificationError(SupersetException):
     """
 
 
+class NotificationTransientError(NotificationError):
+    """Temporary third-party delivery failure surfaced as a report system error."""
+
+
 class SlackV1NotificationError(SupersetException):
     """
     Report should not be run with the slack v1 api

superset/reports/notifications/slack.py

@@ -15,23 +15,19 @@
 # specific language governing permissions and limitations
 # under the License.
 import logging
-from collections.abc import Sequence
-from io import IOBase
-from typing import Union
 
-import backoff
 from flask import g
+from slack_sdk import WebClient
 from slack_sdk.errors import (
     BotUserAccessError,
-    SlackApiError,
     SlackClientConfigurationError,
     SlackClientError,
-    SlackClientNotConnectedError,
     SlackObjectFormationError,
     SlackRequestError,
     SlackTokenRotationError,
 )
 
+from superset import feature_flag_manager
 from superset.reports.models import ReportRecipientType
 from superset.reports.notifications.base import BaseNotification
 from superset.reports.notifications.exceptions import (
@@ -42,23 +38,33 @@ from superset.reports.notifications.exceptions import (
     SlackV1NotificationError,
 )
 from superset.reports.notifications.slack_mixin import SlackMixin
-from superset.utils import json
-from superset.utils.core import recipients_string_to_list
+from superset.reports.notifications.slack_transport import (
+    send_slack_text,
+    send_to_slack_channels,
+)
 from superset.utils.decorators import statsd_gauge
 from superset.utils.slack import (
     get_slack_client,
+    NO_SLACK_RECIPIENTS_MESSAGE,
     should_use_v2_api,
 )
 
 logger = logging.getLogger(__name__)
 
+SLACK_V1_FILE_UPLOAD_MESSAGE = (
+    "Slack v1 file uploads are no longer supported because Slack retired "
+    "`files.upload`. Enable `ALERT_REPORT_SLACK_V2` and grant the Slack bot "
+    "both the `channels:read` and `groups:read` scopes so the recipient can "
+    "be upgraded to Slack v2."
+)
+
 
 # Deprecated: Slack v1 will be removed in the next major release. The Slack
 # `files.upload` endpoint was retired in 2025, so file-bearing sends already
-# fail at the API level; only text-only `chat_postMessage` sends still work
-# here. When the Slack bot has the `channels:read` and `groups:read` scopes,
-# existing v1 recipients are auto-upgraded to SlackV2 on first send via
-# `update_report_schedule_slack_v2`.
+# fail before attempting the retired v1 upload; only text-only
+# `chat_postMessage` sends still work here. When the Slack bot has the
+# `channels:read` and `groups:read` scopes, existing v1 recipients are
+# auto-upgraded to SlackV2 on their first eligible send.
 class SlackNotification(SlackMixin, BaseNotification):  # pylint: disable=too-few-public-methods
     """
     Sends a slack notification for a report recipient
@@ -66,58 +72,42 @@ class SlackNotification(SlackMixin, BaseNotification):  # pylint: disable=too-fe
 
     type = ReportRecipientType.SLACK
 
-    def _get_channel(self) -> str:
-        """
-        Get the recipient's channel(s).
-        Note Slack SDK uses "channel" to refer to one or more
-        channels. Multiple channels are demarcated by a comma.
-        :returns: The comma separated list of channel(s)
-        """
-        recipient_str = json.loads(self._recipient.recipient_config_json)["target"]
+    @staticmethod
+    def _send_text(
+        client: WebClient,
+        channels: list[str],
+        body: str,
+        retry_deadline: float | None = None,
+    ) -> None:
+        """Send a text notification once to each configured channel."""
+        if not channels:
+            raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
+        send_to_slack_channels(
+            channels,
+            lambda target, retry_deadline: send_slack_text(
+                client,
+                target,
+                body,
+                retry_deadline=retry_deadline,
+            ),
+            retry_deadline=retry_deadline,
+        )
 
-        return ",".join(recipients_string_to_list(recipient_str))
+    def _send_legacy_text(self) -> None:
+        if self._content.has_attachments:
+            raise NotificationParamException(SLACK_V1_FILE_UPLOAD_MESSAGE)
 
-    def _get_inline_files(
-        self,
-    ) -> tuple[Union[str, None], Sequence[Union[str, IOBase, bytes]]]:
-        if self._content.csv:
-            return ("csv", [self._content.csv])
-        if self._content.xlsx:
-            return ("xlsx", [self._content.xlsx])
-        if self._content.screenshots:
-            return ("png", self._content.screenshots)
-        if self._content.pdf:
-            return ("pdf", [self._content.pdf])
-        return (None, [])
-
-    @backoff.on_exception(backoff.expo, SlackApiError, factor=10, base=2, max_tries=5)
-    @statsd_gauge("reports.slack.send")
-    def send(self) -> None:
-        file_type, files = self._get_inline_files()
-        title = self._content.name
         body = self._get_body(content=self._content)
         global_logs_context = getattr(g, "logs_context", {}) or {}
-
-        # see if the v2 api will work
-        if should_use_v2_api():
-            # if we can fetch channels, then raise an error and use the v2 api
-            raise SlackV1NotificationError
-
         try:
-            client = get_slack_client()
-            channel = self._get_channel()
-            # files_upload returns SlackResponse as we run it in sync mode.
-            if files:
-                for file in files:
-                    client.files_upload(
-                        channels=channel,
-                        file=file,
-                        initial_comment=body,
-                        title=title,
-                        filetype=file_type,
-                    )
-            else:
-                client.chat_postMessage(channel=channel, text=body)
+            client = get_slack_client(for_delivery=True)
+            channels = self._get_channels()
+            self._send_text(
+                client,
+                channels,
+                body,
+                retry_deadline=self._content.slack_retry_deadline,
+            )
             logger.info(
                 "Report sent to slack",
                 extra={
@@ -134,9 +124,25 @@ class SlackNotification(SlackMixin, BaseNotification):  # pylint: disable=too-fe
             raise NotificationMalformedException(str(ex)) from ex
         except SlackTokenRotationError as ex:
             raise NotificationAuthorizationException(str(ex)) from ex
-        except (SlackClientNotConnectedError, SlackApiError) as ex:
-            raise NotificationUnprocessableException(str(ex)) from ex
         except SlackClientError as ex:
-            # this is the base class for all slack client errors
-            # keep it last so that it doesn't interfere with @backoff
+            # SlackClientError is the base class; keep it last so subclasses
+            # retain their more specific notification classification.
             raise NotificationUnprocessableException(str(ex)) from ex
+
+    @statsd_gauge("reports.slack.send")
+    def send_legacy_text(self) -> None:
+        """Send through Slack v1 without repeating the v2 availability probe."""
+        self._send_legacy_text()
+
+    @statsd_gauge(
+        "reports.slack.send",
+        ignored_exceptions=(SlackV1NotificationError,),
+    )
+    def send(self) -> None:
+        if should_use_v2_api(raise_on_error=self._content.has_attachments):
+            raise SlackV1NotificationError
+        if feature_flag_manager.is_feature_enabled("ALERT_REPORT_SLACK_V2"):
+            # A text-only probe can fail transiently. Still enter the coordinator
+            # so a successful legacy fallback records its warning and metric.
+            raise SlackV1NotificationError
+        self._send_legacy_text()

superset/reports/notifications/slack_channel_resolver.py

@@ -0,0 +1,94 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from superset.reports.notifications.exceptions import NotificationParamException
+from superset.utils.slack import (
+    get_channels_with_search_and_cache_status,
+    refresh_cached_slack_channels_with_search,
+    SlackChannel,
+    SlackChannelTypes,
+)
+
+
+def _match_slack_channel(
+    target: str,
+    channels: list[SlackChannel],
+) -> SlackChannel | None:
+    """Resolve one target with deterministic exact-ID, name, then folded-ID order."""
+    match_groups = (
+        [channel for channel in channels if channel["id"] == target],
+        [
+            channel
+            for channel in channels
+            if channel["name"].casefold() == target.casefold()
+        ],
+        [
+            channel
+            for channel in channels
+            if channel["id"].casefold() == target.casefold()
+        ],
+    )
+    for matches in match_groups:
+        if len(matches) > 1:
+            raise NotificationParamException(
+                f"Slack channel target is ambiguous: {target}"
+            )
+        if matches:
+            return matches[0]
+    return None
+
+
+def _match_slack_channels(
+    targets: list[str],
+    channels: list[SlackChannel],
+) -> tuple[dict[str, SlackChannel], list[str]]:
+    resolved: dict[str, SlackChannel] = {}
+    missing: list[str] = []
+    for target in targets:
+        if channel := _match_slack_channel(target, channels):
+            resolved[target] = channel
+        else:
+            missing.append(target)
+    return resolved, missing
+
+
+def resolve_slack_channel_ids(
+    targets: list[str],
+) -> dict[str, str]:
+    """Resolve Slack names or IDs, refreshing only a stale cached listing."""
+    search_string = ",".join(targets)
+    channels, used_cached_channels = get_channels_with_search_and_cache_status(
+        search_string=search_string,
+        types=[
+            SlackChannelTypes.PRIVATE,
+            SlackChannelTypes.PUBLIC,
+        ],
+        exact_match=True,
+    )
+    channels_by_target, missing_channels = _match_slack_channels(targets, channels)
+    if missing_channels and used_cached_channels:
+        channels = refresh_cached_slack_channels_with_search(
+            search_string=search_string,
+            types=[SlackChannelTypes.PRIVATE, SlackChannelTypes.PUBLIC],
+            exact_match=True,
+        )
+        channels_by_target, missing_channels = _match_slack_channels(targets, channels)
+    if missing_channels:
+        raise NotificationParamException(
+            f"Could not find the following channels: {', '.join(missing_channels)}"
+        )
+    return {target: channel["id"] for target, channel in channels_by_target.items()}

superset/reports/notifications/slack_mixin.py

@@ -18,7 +18,14 @@
 import pandas as pd
 from flask_babel import gettext as __
 
+from superset.reports.models import ReportRecipients
 from superset.reports.notifications.base import NotificationContent
+from superset.reports.notifications.exceptions import NotificationParamException
+from superset.utils import json
+from superset.utils.slack import (
+    NO_SLACK_RECIPIENTS_MESSAGE,
+    parse_slack_recipient_targets,
+)
 
 # Slack only allows Markdown messages up to 4k chars
 MAXIMUM_MESSAGE_SIZE = 4000
@@ -26,6 +33,20 @@ MAXIMUM_MESSAGE_SIZE = 4000
 
 # pylint: disable=too-few-public-methods
 class SlackMixin:
+    _recipient: ReportRecipients
+
+    def _get_channels(self) -> list[str]:
+        """Return normalized Slack targets without duplicates."""
+        try:
+            recipient_str = json.loads(self._recipient.recipient_config_json)["target"]
+        except (KeyError, TypeError, ValueError) as ex:
+            raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE) from ex
+
+        if not isinstance(recipient_str, str):
+            raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
+
+        return parse_slack_recipient_targets(recipient_str)
+
     def _message_template(
         self,
         content: NotificationContent,

superset/reports/notifications/slack_transport.py

@@ -0,0 +1,310 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import math
+import time
+from collections.abc import Callable
+from functools import partial
+from typing import TypeVar
+from urllib.error import HTTPError
+
+import backoff
+from flask import current_app as app
+from slack_sdk import WebClient
+from slack_sdk.errors import SlackApiError, SlackRequestError
+
+from superset.reports.notifications.exceptions import (
+    NotificationTransientError,
+    NotificationUnprocessableException,
+)
+from superset.utils.slack import (
+    get_slack_api_error_code,
+    get_slack_api_status_code,
+    is_retryable_slack_transport_error,
+    is_transient_slack_api_error,
+    is_transient_slack_transport_error,
+    SLACK_TRANSIENT_TRANSPORT_ERRORS,
+)
+
+SLACK_API_TIMEOUT_MARGIN = 1
+
+_SlackApiResult = TypeVar("_SlackApiResult")
+
+_SLACK_RETRY_DEADLINE_MESSAGE = (
+    "Slack send retry deadline exceeded; increase SLACK_SEND_RETRY_MAX_TIME "
+    "for slow Slack workspaces or large-file reports"
+)
+
+
+class SlackRetryDeadlineError(Exception):
+    """A Slack operation was skipped because the shared send budget expired."""
+
+    def __init__(self) -> None:
+        super().__init__(_SLACK_RETRY_DEADLINE_MESSAGE)
+
+
+class SlackChannelResponseError(SlackRequestError):
+    """Slack returned malformed channel-specific data before a terminal write."""
+
+
+_SLACK_RETRY_ERRORS = (SlackApiError, *SLACK_TRANSIENT_TRANSPORT_ERRORS)
+_SLACK_CHANNEL_FAILURES = (
+    *_SLACK_RETRY_ERRORS,
+    SlackChannelResponseError,
+    SlackRetryDeadlineError,
+)
+
+
+def _get_slack_send_retry_max_time() -> float:
+    """Return the effective schedule-wide Slack send budget in seconds."""
+    configured_budget = float(app.config.get("SLACK_SEND_RETRY_MAX_TIME", 150))
+    request_timeout = float(app.config.get("SLACK_API_TIMEOUT", 30))
+    return max(
+        configured_budget,
+        request_timeout + SLACK_API_TIMEOUT_MARGIN,
+    )
+
+
+def get_slack_send_retry_deadline(max_deadline: float | None = None) -> float:
+    """Return one configured Slack deadline, optionally clamped by its caller."""
+    configured_deadline = time.monotonic() + _get_slack_send_retry_max_time()
+    if max_deadline is None:
+        return configured_deadline
+    return min(max_deadline, configured_deadline)
+
+
+def get_slack_request_timeout(
+    client_timeout: int | float,
+    retry_deadline: float,
+) -> int:
+    """Clamp one Slack request timeout to the remaining delivery budget."""
+    remaining = retry_deadline - time.monotonic()
+    if remaining <= 0:
+        raise SlackRetryDeadlineError
+    request_timeout = int(min(float(client_timeout), remaining))
+    if request_timeout <= 0:
+        raise SlackRetryDeadlineError
+    return request_timeout
+
+
+def _give_up_slack_api_retry(
+    ex: Exception,
+    *,
+    retry_transient_errors: bool = True,
+    retry_transport_errors: bool = False,
+) -> bool:
+    """Return whether application backoff should stop retrying a Slack call."""
+    if isinstance(ex, HTTPError) and ex.code == 429:
+        return True
+    if not isinstance(ex, SlackApiError):
+        if not retry_transient_errors:
+            return True
+        return not (
+            is_transient_slack_transport_error(ex)
+            if retry_transport_errors
+            else is_retryable_slack_transport_error(ex)
+        )
+
+    status_code = get_slack_api_status_code(ex)
+    # call_slack_api handles HTTP 429 within the shared monotonic deadline.
+    # Retrying an exhausted 429 through the outer backoff would multiply the
+    # operator-configured rate-limit budget.
+    error_code = get_slack_api_error_code(ex)
+    if status_code == 429:
+        return True
+    if not retry_transient_errors:
+        return True
+    if is_transient_slack_api_error(ex, error_code):
+        return False
+    if status_code is not None and 400 <= status_code < 500:
+        return True
+    return bool(error_code)
+
+
+def _get_slack_retry_after(ex: SlackApiError | HTTPError) -> float | None:
+    response = getattr(ex, "response", ex)
+    headers = getattr(response, "headers", None)
+    if headers is None:
+        return None
+    for name in headers.keys():
+        if name.lower() != "retry-after":
+            continue
+        value = headers.get(name)
+        raw_value = value[0] if isinstance(value, list) else value
+        try:
+            retry_after = float(raw_value)
+            return max(retry_after, 0.0) if math.isfinite(retry_after) else None
+        except (TypeError, ValueError):
+            return None
+    return None
+
+
+def _get_slack_rate_limit_status(ex: SlackApiError | HTTPError) -> int | None:
+    if isinstance(ex, HTTPError):
+        return ex.code
+    return get_slack_api_status_code(ex)
+
+
+def call_slack_api(
+    method: Callable[..., _SlackApiResult],
+    *,
+    retry_deadline: float | None = None,
+    retry_transient_errors: bool = True,
+    retry_transport_errors: bool = False,
+    retry_rate_limits: bool = True,
+    **kwargs: object,
+) -> _SlackApiResult:
+    """Call Slack with bounded retries, optionally sharing an outer deadline."""
+    if retry_deadline is None:
+        retry_deadline = time.monotonic() + _get_slack_send_retry_max_time()
+    max_time = retry_deadline - time.monotonic()
+    if max_time <= 0:
+        raise SlackRetryDeadlineError
+
+    max_rate_limit_retries = (
+        max(
+            int(app.config.get("SLACK_API_RATE_LIMIT_RETRY_COUNT", 2)),
+            0,
+        )
+        if retry_rate_limits
+        else 0
+    )
+    rate_limit_retries = 0
+
+    @backoff.on_exception(
+        backoff.expo,
+        _SLACK_RETRY_ERRORS,
+        factor=10,
+        base=2,
+        max_tries=5,
+        max_time=max_time,
+        giveup=partial(
+            _give_up_slack_api_retry,
+            retry_transient_errors=retry_transient_errors,
+            retry_transport_errors=retry_transport_errors,
+        ),
+    )
+    def call() -> _SlackApiResult:
+        nonlocal rate_limit_retries
+        while True:
+            if time.monotonic() >= retry_deadline:
+                raise SlackRetryDeadlineError
+            try:
+                return method(**kwargs)
+            except (SlackApiError, HTTPError) as ex:
+                if (
+                    _get_slack_rate_limit_status(ex) != 429
+                    or rate_limit_retries >= max_rate_limit_retries
+                ):
+                    raise
+                retry_after = _get_slack_retry_after(ex)
+                if retry_after is None:
+                    raise
+                remaining = retry_deadline - time.monotonic()
+                if retry_after >= remaining:
+                    raise SlackRetryDeadlineError from ex
+                time.sleep(retry_after)
+                rate_limit_retries += 1
+
+    return call()
+
+
+def call_slack_api_with_timeout(
+    client: WebClient,
+    method: Callable[..., _SlackApiResult],
+    *,
+    retry_deadline: float,
+    retry_transient_errors: bool = True,
+    retry_transport_errors: bool = False,
+    retry_rate_limits: bool = True,
+    **kwargs: object,
+) -> _SlackApiResult:
+    """Call Slack with the SDK request timeout capped by the shared budget."""
+    original_timeout = client.timeout
+
+    def call() -> _SlackApiResult:
+        client.timeout = get_slack_request_timeout(original_timeout, retry_deadline)
+        try:
+            return method(**kwargs)
+        finally:
+            client.timeout = original_timeout
+
+    return call_slack_api(
+        call,
+        retry_deadline=retry_deadline,
+        retry_transient_errors=retry_transient_errors,
+        retry_transport_errors=retry_transport_errors,
+        retry_rate_limits=retry_rate_limits,
+    )
+
+
+def send_slack_text(
+    client: WebClient,
+    channel: str,
+    text: str,
+    retry_deadline: float,
+) -> None:
+    """Post one Slack text message without replaying an ambiguous terminal write."""
+    call_slack_api_with_timeout(
+        client,
+        client.chat_postMessage,
+        retry_deadline=retry_deadline,
+        retry_transient_errors=False,
+        channel=channel,
+        text=text,
+    )
+
+
+def _is_transient_slack_channel_failure(error: Exception) -> bool:
+    """Return whether any failed destination requires retrying the report."""
+    return bool(
+        isinstance(error, SlackRetryDeadlineError)
+        or isinstance(error, SlackChannelResponseError)
+        or is_transient_slack_transport_error(error)
+        or (
+            isinstance(error, SlackApiError)
+            and is_transient_slack_api_error(
+                error,
+                get_slack_api_error_code(error),
+            )
+        )
+    )
+
+
+def send_to_slack_channels(
+    channels: list[str],
+    send_to_channel: Callable[[str, float], None],
+    *,
+    retry_deadline: float | None = None,
+) -> None:
+    """Send to each channel within one schedule-wide application deadline."""
+    retry_deadline = get_slack_send_retry_deadline(retry_deadline)
+    failures: list[tuple[str, Exception]] = []
+    for channel in channels:
+        try:
+            send_to_channel(channel, retry_deadline)
+        except _SLACK_CHANNEL_FAILURES as ex:
+            failures.append((channel, ex))
+
+    if not failures:
+        return
+
+    details = "; ".join(f"{channel}: {error}" for channel, error in failures)
+    message = f"Slack delivery failed for the following channels: {details}"
+    if any(_is_transient_slack_channel_failure(error) for _, error in failures):
+        raise NotificationTransientError(message) from failures[0][1]
+    raise NotificationUnprocessableException(message) from failures[0][1]

superset/reports/notifications/slackv2.py

@@ -15,18 +15,25 @@
 # specific language governing permissions and limitations
 # under the License.
 import logging
-from collections.abc import Callable, Sequence
-from io import IOBase
-from typing import List, Union
+from contextlib import closing
+from email.message import Message
+from ssl import SSLContext
+from urllib.error import HTTPError
+from urllib.parse import urlparse
+from urllib.request import (
+    build_opener,
+    HTTPSHandler,
+    ProxyHandler,
+    Request,
+    urlopen,
+)
 
-import backoff
 from flask import g
+from slack_sdk import WebClient
 from slack_sdk.errors import (
     BotUserAccessError,
-    SlackApiError,
     SlackClientConfigurationError,
     SlackClientError,
-    SlackClientNotConnectedError,
     SlackObjectFormationError,
     SlackRequestError,
     SlackTokenRotationError,
@@ -41,61 +48,120 @@ from superset.reports.notifications.exceptions import (
     NotificationUnprocessableException,
 )
 from superset.reports.notifications.slack_mixin import SlackMixin
-from superset.utils import json
-from superset.utils.core import recipients_string_to_list
+from superset.reports.notifications.slack_transport import (
+    call_slack_api,
+    call_slack_api_with_timeout,
+    get_slack_request_timeout,
+    send_slack_text,
+    send_to_slack_channels,
+    SlackChannelResponseError,
+)
 from superset.utils.decorators import statsd_gauge
-from superset.utils.slack import get_slack_client
+from superset.utils.slack import (
+    get_slack_client,
+    NO_SLACK_RECIPIENTS_MESSAGE,
+)
 
 logger = logging.getLogger(__name__)
 
-_TRANSIENT_SLACK_API_ERROR_CODES = frozenset(
-    {
-        "fatal_error",
-        "internal_error",
-        "ratelimited",
-        "request_timeout",
-        "rollup_error",
-        "service_unavailable",
-        "timeout",
-    }
-)
+
+def _upload_file_data(
+    *,
+    url: str,
+    data: bytes,
+    timeout: int,
+    proxy: str | None,
+    ssl: SSLContext | None,
+) -> tuple[int, str]:
+    """Upload bytes to Slack's issued URL using stable stdlib HTTP APIs."""
+    if urlparse(url).scheme != "https":
+        raise SlackRequestError("Slack upload URL must use HTTPS")
+    request = Request(method="POST", url=url, data=data)  # noqa: S310
+    if proxy is not None:
+        if not isinstance(proxy, str):
+            raise SlackRequestError(
+                f"Invalid proxy detected: {proxy} must be a str value"
+            )
+        response = build_opener(
+            ProxyHandler({"http": proxy, "https": proxy}),
+            HTTPSHandler(context=ssl),
+        ).open(request, timeout=timeout)
+    else:
+        response = urlopen(request, context=ssl, timeout=timeout)  # noqa: S310
+
+    with closing(response):
+        charset = response.headers.get_content_charset() or "utf-8"
+        body = response.read().decode(charset)
+        return response.status, body
 
 
-def _get_slack_api_error_code(ex: SlackApiError) -> str:
-    response = getattr(ex, "response", None)
-    data = getattr(response, "data", None)
-    if not isinstance(data, dict):
-        data = response if isinstance(response, dict) else {}
-    return str(data.get("error") or "")
+def _upload_file_to_slack(
+    client: WebClient,
+    *,
+    channel: str,
+    file: bytes,
+    initial_comment: str,
+    title: str,
+    filename: str,
+    retry_deadline: float,
+) -> None:
+    """Upload one file without replaying completed phases during retries."""
+    data = file
+    upload_url_response = call_slack_api_with_timeout(
+        client,
+        client.files_getUploadURLExternal,
+        retry_deadline=retry_deadline,
+        retry_transport_errors=True,
+        filename=filename,
+        length=len(data),
+    )
+    try:
+        file_id = upload_url_response.get("file_id")
+        upload_url = upload_url_response.get("upload_url")
+    except (AttributeError, TypeError) as ex:
+        raise SlackChannelResponseError(
+            "Slack did not return valid upload metadata"
+        ) from ex
+    if (
+        not isinstance(file_id, str)
+        or not file_id
+        or not isinstance(upload_url, str)
+        or not upload_url
+    ):
+        raise SlackChannelResponseError("Slack did not return a file ID and upload URL")
 
+    def upload_file() -> None:
+        timeout = get_slack_request_timeout(client.timeout, retry_deadline)
+        status, response_body = _upload_file_data(
+            url=upload_url,
+            data=data,
+            timeout=timeout,
+            proxy=client.proxy,
+            ssl=client.ssl,
+        )
+        if status != 200:
+            raise HTTPError(
+                upload_url,
+                status,
+                f"Slack external upload failed: {response_body}",
+                Message(),
+                None,
+            )
 
-def _get_slack_api_status_code(ex: SlackApiError) -> int | None:
-    response = getattr(ex, "response", None)
-    return getattr(response, "status_code", None)
-
-
-def _give_up_slack_api_retry(ex: Exception) -> bool:
-    if not isinstance(ex, SlackApiError):
-        return False
-
-    status_code = _get_slack_api_status_code(ex)
-    if status_code == 429 or (status_code is not None and 500 <= status_code < 600):
-        return False
-
-    error_code = _get_slack_api_error_code(ex)
-    return bool(error_code and error_code not in _TRANSIENT_SLACK_API_ERROR_CODES)
-
-
-@backoff.on_exception(
-    backoff.expo,
-    (SlackApiError, SlackClientNotConnectedError),
-    factor=10,
-    base=2,
-    max_tries=5,
-    giveup=_give_up_slack_api_retry,
-)
-def _call_slack_api(method: Callable[..., object], **kwargs: object) -> None:
-    method(**kwargs)
+    call_slack_api(
+        upload_file,
+        retry_deadline=retry_deadline,
+        retry_transport_errors=True,
+    )
+    call_slack_api_with_timeout(
+        client,
+        client.files_completeUploadExternal,
+        retry_deadline=retry_deadline,
+        retry_transient_errors=False,
+        files=[{"id": file_id, "title": title}],
+        channel_id=channel,
+        initial_comment=initial_comment,
+    )
 
 
 class SlackV2Notification(SlackMixin, BaseNotification):  # pylint: disable=too-few-public-methods
@@ -105,19 +171,9 @@ class SlackV2Notification(SlackMixin, BaseNotification):  # pylint: disable=too-
 
     type = ReportRecipientType.SLACKV2
 
-    def _get_channels(self) -> List[str]:
-        """
-        Get the recipient's channel(s).
-        :returns: A list of channel ids: "EID676L"
-        :raises NotificationParamException or SlackApiError: If the recipient is not found
-        """  # noqa: E501
-        recipient_str = json.loads(self._recipient.recipient_config_json)["target"]
-
-        return recipients_string_to_list(recipient_str)
-
     def _get_inline_files(
         self,
-    ) -> tuple[Union[str, None], Sequence[Union[str, IOBase, bytes]]]:
+    ) -> tuple[str | None, list[bytes]]:
         if self._content.csv:
             return ("csv", [self._content.csv])
         if self._content.xlsx:
@@ -132,24 +188,28 @@ class SlackV2Notification(SlackMixin, BaseNotification):  # pylint: disable=too-
     def send(self) -> None:
         global_logs_context = getattr(g, "logs_context", {}) or {}
         try:
-            client = get_slack_client()
+            client = get_slack_client(for_delivery=True)
             title = self._content.name
             body = self._get_body(content=self._content)
 
             channels = self._get_channels()
 
             if not channels:
-                raise NotificationParamException("No recipients saved in the report")
+                raise NotificationParamException(NO_SLACK_RECIPIENTS_MESSAGE)
 
             file_type, files = self._get_inline_files()
-            file_name = f"{title}.{file_type}"
 
-            # files_upload returns SlackResponse as we run it in sync mode.
-            for channel in channels:
+            def send_to_channel(channel: str, retry_deadline: float) -> None:
                 if len(files) > 0:
+                    if file_type is None:
+                        raise SlackChannelResponseError(
+                            "Slack upload file type was not provided"
+                        )
+                    file_name = f"{title}.{file_type}"
                     for file in files:
-                        _call_slack_api(
-                            client.files_upload_v2,
+                        _upload_file_to_slack(
+                            client,
+                            retry_deadline=retry_deadline,
                             channel=channel,
                             file=file,
                             initial_comment=body,
@@ -157,7 +217,18 @@ class SlackV2Notification(SlackMixin, BaseNotification):  # pylint: disable=too-
                             filename=file_name,
                         )
                 else:
-                    _call_slack_api(client.chat_postMessage, channel=channel, text=body)
+                    send_slack_text(
+                        client,
+                        channel,
+                        body,
+                        retry_deadline=retry_deadline,
+                    )
+
+            send_to_slack_channels(
+                channels,
+                send_to_channel,
+                retry_deadline=self._content.slack_retry_deadline,
+            )
 
             logger.info(
                 "Report sent to slack",
@@ -175,9 +246,7 @@ class SlackV2Notification(SlackMixin, BaseNotification):  # pylint: disable=too-
             raise NotificationMalformedException(str(ex)) from ex
         except SlackTokenRotationError as ex:
             raise NotificationAuthorizationException(str(ex)) from ex
-        except (SlackClientNotConnectedError, SlackApiError) as ex:
-            raise NotificationUnprocessableException(str(ex)) from ex
         except SlackClientError as ex:
-            # this is the base class for all slack client errors
-            # keep it last so that it doesn't interfere with @backoff
+            # SlackClientError is the base class; keep it last so subclasses
+            # retain their more specific notification classification.
             raise NotificationUnprocessableException(str(ex)) from ex

superset/security/manager.py

@@ -38,6 +38,7 @@ from flask import current_app, Flask, g, has_app_context, Request, Response
 from flask_appbuilder import Model
 from flask_appbuilder.api import expose, protect, safe
 from flask_appbuilder.models.filters import BaseFilter
+from flask_appbuilder.security.manager import AUTH_REMOTE_USER
 from flask_appbuilder.security.sqla.apis import GroupApi, RoleApi, UserApi
 from flask_appbuilder.security.sqla.apis.permission_view_menu.api import (
     PermissionViewMenuApi,
@@ -3090,7 +3091,7 @@ class SupersetSecurityManager(  # pylint: disable=too-many-public-methods
             logger.warning(
                 "Dataset has no database will retry with database_id to set permission"
             )
-            database = self.session.query(Database).get(target.database_id)
+            database = self.session.get(Database, target.database_id)
             dataset_perm = self.get_dataset_perm(
                 target.id, target.table_name, database.database_name
             )
@@ -3971,6 +3972,21 @@ class SupersetSecurityManager(  # pylint: disable=too-many-public-methods
             if query in self.session:
                 self.session.expunge(query)
 
+        # When only ``database`` is provided, enforce database-level access
+        # here so the call is not a no-op.
+        if database and not (table or query):
+            if not self.can_access_database(database):
+                raise SupersetSecurityException(
+                    SupersetError(
+                        error_type=SupersetErrorType.DATABASE_SECURITY_ACCESS_ERROR,
+                        message=_(
+                            "You need access to the following database: %(name)s",
+                            name=database.database_name,
+                        ),
+                        level=ErrorLevel.WARNING,
+                    )
+                )
+
         if database and table or query:
             if query:
                 # Type narrow: only SQL Lab Query objects have .database attribute
@@ -4053,6 +4069,24 @@ class SupersetSecurityManager(  # pylint: disable=too-many-public-methods
                             level=ErrorLevel.ERROR,
                         )
                     )
+                # Statements that rebind how unqualified table names resolve
+                # (``USE``, ``SET SCHEMA``, or a ``search_path`` change) make
+                # the qualification below diverge from what the engine uses at
+                # execution time, so reject them regardless of engine.
+                if force_dataset_match and parse_result.script.changes_default_schema():
+                    raise SupersetSecurityException(
+                        SupersetError(
+                            error_type=SupersetErrorType.QUERY_SECURITY_ACCESS_ERROR,
+                            message=_(
+                                "SQL Lab cannot authorise a script that "
+                                "changes the schema used to resolve "
+                                "unqualified table names (e.g. USE or "
+                                "search_path changes). Qualify tables "
+                                "explicitly instead."
+                            ),
+                            level=ErrorLevel.ERROR,
+                        )
+                    )
                 tables = {
                     table_.qualify(
                         catalog=query.catalog or default_catalog,
@@ -4337,6 +4371,25 @@ class SupersetSecurityManager(  # pylint: disable=too-many-public-methods
             elif chart.datasource and self.can_access_datasource(chart.datasource):
                 return
 
+            # An embedded guest may access a member chart of a dashboard their
+            # guest token grants. Embedded dashboards render their member charts
+            # client-side, so the chart definitions must be served even though a
+            # guest holds no standalone datasource grant. The chart's dataset
+            # must still satisfy any allowlist the token carries, and data
+            # queries are re-checked through the datasource branch above (which
+            # receives the dashboard context in the chart-data form_data).
+            if (
+                is_feature_enabled("EMBEDDED_SUPERSET")
+                and self.is_guest_user()
+                and any(
+                    self.has_guest_access(dashboard_) for dashboard_ in chart.dashboards
+                )
+                and self._guest_token_allows_dataset(
+                    chart.datasource.id if chart.datasource else None
+                )
+            ):
+                return
+
             raise SupersetSecurityException(self.get_chart_access_error_object(chart))
 
     def get_user_by_username(self, username: str) -> Optional[User]:
@@ -4894,6 +4947,26 @@ class SupersetSecurityManager(  # pylint: disable=too-many-public-methods
             return user
         return None
 
+    def _guest_token_allows_dataset(self, datasource_id: Optional[int]) -> bool:
+        """Return whether the current guest token permits this dataset.
+
+        A token without a ``datasets`` allowlist permits every dataset
+        (backward compatible). A token that carries one permits only the listed
+        integer IDs; a malformed allowlist permits nothing. Non-guest callers
+        are unaffected: they hold no guest token and always get ``True``.
+        """
+        guest_user = self.get_current_guest_user_if_guest()
+        if not guest_user:
+            return True
+        allowed_datasets: Optional[list[int]] = guest_user.guest_token.get("datasets")
+        if allowed_datasets is None:
+            return True
+        return (
+            isinstance(allowed_datasets, list)
+            and all(isinstance(d, int) for d in allowed_datasets)
+            and datasource_id in allowed_datasets
+        )
+
     def has_guest_access(self, dashboard: "Dashboard") -> bool:
         user = self.get_current_guest_user_if_guest()
         if not user:
@@ -5067,7 +5140,19 @@ class SupersetSecurityManager(  # pylint: disable=too-many-public-methods
     def register_views(self) -> None:
         from superset.views.auth import SupersetAuthView, SupersetRegisterUserView
 
-        if self.register_superset_auth_view:
+        # AUTH_REMOTE_USER has no interactive login form to render: the whole
+        # point is that an upstream proxy already authenticated the request and
+        # passes the identity via a header/env var, so FlaskAppBuilder's
+        # AuthRemoteUserView performs a silent GET-time login with no UI. That
+        # view registers at the same "/login/" route as SupersetAuthView, and
+        # since both add distinct Flask endpoints for the same URL rule,
+        # whichever gets registered first wins the dispatch -- SupersetAuthView
+        # always wins because it's added before super().register_views() runs
+        # FlaskAppBuilder's own auth_type dispatch. That silently shadows
+        # AUTH_REMOTE_USER: the SPA login shell renders instead of the header
+        # ever being checked. Skip registering it for this auth type so
+        # FlaskAppBuilder's AuthRemoteUserView actually claims the route.
+        if self.register_superset_auth_view and self.auth_type != AUTH_REMOTE_USER:
             self.auth_view = self.appbuilder.add_view_no_menu(SupersetAuthView)
         if self.register_superset_registeruser_view:
             self.registeruser_view = self.appbuilder.add_view_no_menu(

superset/sql/parse.py

@@ -275,6 +275,9 @@ class RLSTransformer:
 
         return None
 
+    def __call__(self, node: exp.Table) -> exp.Expression:
+        raise NotImplementedError()
+
 
 class RLSAsPredicateTransformer(RLSTransformer):
     """
@@ -298,17 +301,17 @@ class RLSAsPredicateTransformer(RLSTransformer):
     databases without support for subqueries.
     """
 
-    def __call__(self, node: exp.Expression) -> exp.Expression:
-        if not isinstance(node, exp.Table):
-            return node
-
+    def __call__(self, node: exp.Table) -> exp.Expression:
         predicate = self.get_predicate(node)
         if not predicate:
             return node
 
-        # qualify columns with table name
+        # Qualify with the parsed alias node, not the ``node.alias`` string (which drops
+        # quoting and could inject SQL); use the table when the alias has no name.
+        table_alias = node.args.get("alias")
+        qualifier = (table_alias and table_alias.this) or node.this
         for column in predicate.find_all(exp.Column):
-            column.set("table", node.alias or node.this)
+            column.set("table", qualifier.copy())
 
         if isinstance(node.parent, exp.From):
             select = node.parent.parent
@@ -354,13 +357,12 @@ class RLSAsSubqueryTransformer(RLSTransformer):
     all databases.
     """
 
-    def __call__(self, node: exp.Expression) -> exp.Expression:
-        if not isinstance(node, exp.Table):
-            return node
-
+    def __call__(self, node: exp.Table) -> exp.Expression:
         if predicate := self.get_predicate(node):
-            if node.alias:
-                alias = node.alias
+            if existing_alias := node.args.get("alias"):
+                # Reuse the parsed alias node, not the ``node.alias`` string: that drops
+                # quoting (SQL in an alias re-emits as SQL) and the column-alias list.
+                alias = existing_alias
             else:
                 # Use just the table name (not schema-qualified) so that
                 # column references like ``table.column`` still resolve after
@@ -616,6 +618,18 @@ class BaseSQLStatement(Generic[InternalRepresentation]):
         """
         return False
 
+    def changes_default_schema(self) -> bool:
+        """
+        Check if the statement changes the schema used to resolve unqualified
+        table names.
+
+        Defaults to ``False``; engines whose statements can rebind unqualified
+        schema resolution override this.
+
+        :return: True if the statement rebinds default schema resolution
+        """
+        return False
+
     def get_disallowed_tables(
         self,
         tables: set[str],
@@ -751,26 +765,24 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
         }
     )
 
-    # PostgreSQL constructs that sqlglot represents as an opaque ``exp.Command``
-    # (no structured AST). Each can mutate server state or wrap a DML body that
-    # would otherwise be detected by node-type matching. Used by
-    # ``is_mutating()``.
-    _POSTGRES_MUTATING_COMMAND_NAMES: frozenset[str] = frozenset(
+    # Constructs that sqlglot represents as an opaque ``exp.Command`` (no
+    # structured AST). Each can mutate server state or wrap a DML body that
+    # would otherwise be detected by node-type matching. The head keywords
+    # are not engine-specific (MySQL ``CALL`` / ``LOAD DATA INFILE`` and
+    # MSSQL ``EXEC`` reach the same ``exp.Command`` fallback as their
+    # PostgreSQL counterparts), so ``is_mutating()`` applies this list for
+    # every dialect: an opaque command with one of these heads is treated as
+    # mutating.
+    _MUTATING_COMMAND_NAMES: frozenset[str] = frozenset(
         {
             "DO",  # PL/pgSQL anonymous block
             "PREPARE",  # PREPARE u AS UPDATE ... ; EXECUTE u
             "EXECUTE",  # body is the prepared DML
+            "EXEC",  # MSSQL spelling of EXECUTE; the procedure body may mutate
             "CALL",  # procedure body may mutate
             "COPY",  # server-side file ingest into a table
             "GRANT",
             "REVOKE",
-            # Only the command-fallback forms (e.g. SET ROLE / SET SESSION
-            # AUTHORIZATION, which change the effective user) reach here as an
-            # exp.Command. Structured `SET search_path = ...` /
-            # `SET statement_timeout = ...` parse as exp.Set and are NOT matched
-            # by this command-name path.
-            "SET",
-            "RESET",  # RESET ROLE / RESET ALL reverts SET; same class as SET
             "REFRESH",  # REFRESH MATERIALIZED VIEW
             "REINDEX",
             "VACUUM",
@@ -783,7 +795,9 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
             "CREATE",
             "ALTER",
             "DROP",
-            "LOAD",  # LOAD '/path/lib.so' dlopens a shared library on the PG host
+            # MySQL LOAD DATA INFILE ingests server files into a table;
+            # PostgreSQL LOAD '/path/lib.so' dlopens a shared library.
+            "LOAD",
             # NOTE: `SHOW` is intentionally NOT included. It is a read (mutates
             # nothing), so classifying it as mutating would be wrong for every
             # is_mutating()/has_mutation() consumer (the commit decision, the
@@ -794,6 +808,20 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
         }
     )
 
+    # PostgreSQL-only command-fallback heads. Only the command-fallback
+    # forms (e.g. SET ROLE / SET SESSION AUTHORIZATION, which change the
+    # effective user) reach here as an exp.Command; structured
+    # `SET search_path = ...` / `SET statement_timeout = ...` parse as
+    # exp.Set and are NOT matched by this path. On other dialects the `SET`
+    # fallback covers session variables (e.g. Hive `SET hivevar:x=1`),
+    # which do not mutate data, so these heads stay dialect-gated.
+    _POSTGRES_MUTATING_COMMAND_NAMES: frozenset[str] = frozenset(
+        {
+            "SET",
+            "RESET",  # RESET ROLE / RESET ALL reverts SET; same class as SET
+        }
+    )
+
     # Dialects where `SELECT ... INTO target` is CTAS (creates a table, and so
     # mutates schema). Elsewhere the same syntax assigns into a variable and is
     # a read: Oracle PL/SQL `SELECT ... INTO v` and MySQL `SELECT ... INTO @v`
@@ -926,7 +954,7 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
         """
         return isinstance(self._parsed, exp.Select)
 
-    def is_mutating(self) -> bool:
+    def is_mutating(self) -> bool:  # noqa: C901
         """
         Check if the statement mutates data (DDL/DML).
 
@@ -949,6 +977,14 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
             exp.Revoke,
             # COMMENT ON TABLE/COLUMN/etc. writes to system catalog pg_description.
             exp.Comment,
+            # A bare COMMIT persists earlier writes on the same connection, so
+            # treat it as mutating.
+            exp.Commit,
+            # EXEC/EXECUTE invokes a stored procedure whose body is opaque;
+            # some dialects (e.g. MSSQL) parse it as this structured node
+            # rather than an opaque exp.Command, so treat it as mutating here
+            # too.
+            exp.Execute,
         )
 
         if self._parsed.find(*mutating_nodes):
@@ -986,37 +1022,76 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
         ):
             return True
 
-        # depending on the dialect (Oracle, MS SQL) the `ALTER` is parsed as a
-        # command, not an expression - check at root level
-        if isinstance(self._parsed, exp.Command) and self._parsed.name == "ALTER":
-            return True  # pragma: no cover
+        # Statements that sqlglot cannot model parse as an opaque
+        # `exp.Command`. The `.name` attribute on `exp.Command` preserves
+        # the source-case of the head keyword (so `create extension ...`
+        # would yield `'create'`), which means the lookups must be
+        # case-insensitive. This also covers the dialects (Oracle, MS SQL)
+        # where `ALTER` itself is parsed as a command, not an expression.
+        if isinstance(self._parsed, exp.Command):
+            command_name = self._parsed.name.upper()
 
-        # PostgreSQL constructs that sqlglot represents as an opaque
-        # `exp.Command` rather than a structured AST. Each of these can mutate
-        # state or wrap a DML body that would otherwise be detected. The
-        # `.name` attribute on `exp.Command` preserves the source-case of the
-        # head keyword (so `create extension ...` would yield `'create'`),
-        # which means the set lookup must be case-insensitive.
-        if (
-            self._dialect == Dialects.POSTGRES
-            and isinstance(self._parsed, exp.Command)
-            and self._parsed.name.upper() in self._POSTGRES_MUTATING_COMMAND_NAMES
-        ):
-            return True
+            if command_name in self._MUTATING_COMMAND_NAMES:
+                return True
 
-        # Postgres runs DMLs prefixed by `EXPLAIN ANALYZE`, see
-        # https://www.postgresql.org/docs/current/sql-explain.html
-        if (
-            self._dialect == Dialects.POSTGRES
-            and isinstance(self._parsed, exp.Command)
-            and self._parsed.name == "EXPLAIN"
-            and self._parsed.expression.name.upper().startswith("ANALYZE ")
-        ):
-            analyzed_sql = self._parsed.expression.name[len("ANALYZE ") :]
-            return SQLStatement(
-                statement=analyzed_sql,
-                engine=self.engine,
-            ).is_mutating()
+            if (
+                self._dialect == Dialects.POSTGRES
+                and command_name in self._POSTGRES_MUTATING_COMMAND_NAMES
+            ):
+                return True
+
+            # `EXPLAIN ANALYZE <statement>` executes the statement for real
+            # (PostgreSQL and MySQL both run the body), see
+            # https://www.postgresql.org/docs/current/sql-explain.html
+            # The flag may be spelled `ANALYSE`, be separated by any
+            # whitespace, or appear in a parenthesized option list such as
+            # `EXPLAIN (ANALYZE, BUFFERS) ...`, so the raw tail is
+            # normalized before the inner statement is classified. Anything
+            # that carries the flag but cannot be classified is treated as
+            # mutating.
+            if command_name == "EXPLAIN":
+                tail = (
+                    self._parsed.expression.name.strip()
+                    if self._parsed.expression
+                    else ""
+                )
+
+                # sqlglot preserves the raw tail text, comments included;
+                # strip leading comments so an option list hidden behind
+                # `/* ... */` or `-- ...` is still recognized.
+                while True:
+                    if tail.startswith("/*") and "*/" in tail:
+                        tail = tail.split("*/", 1)[1].lstrip()
+                    elif tail.startswith("--"):
+                        parts = tail.split("\n", 1)
+                        tail = parts[1].lstrip() if len(parts) > 1 else ""
+                    else:
+                        break
+
+                has_analyze = False
+                if tail.startswith("("):
+                    options, _, tail = tail[1:].partition(")")
+                    has_analyze = bool(
+                        re.search(r"\b(ANALYZE|ANALYSE)\b", options, re.IGNORECASE)
+                    )
+                else:
+                    while match := re.match(
+                        r"(ANALYZE|ANALYSE|VERBOSE)\s+", tail, re.IGNORECASE
+                    ):
+                        if match.group(1).upper() != "VERBOSE":
+                            has_analyze = True
+                        tail = tail[match.end() :]
+
+                if has_analyze:
+                    if not (inner_sql := tail.strip()):
+                        return True
+                    try:
+                        return SQLStatement(
+                            statement=inner_sql,
+                            engine=self.engine,
+                        ).is_mutating()
+                    except SupersetParseError:
+                        return True
 
         return False
 
@@ -1188,6 +1263,58 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
             return bool(tokens) and tokens[0].strip('"').lower() == "search_path"
         return False
 
+    def changes_default_schema(self) -> bool:
+        """
+        Return True if the statement rebinds default schema resolution.
+
+        Covers ``USE`` statements (MySQL-, Doris- and Snowflake-family
+        engines) and ``SET [CURRENT] SCHEMA`` / ``SET CATALOG`` variants, in
+        addition to anything that changes the Postgres ``search_path``.
+        Unqualified table names in later statements on the same cursor then
+        resolve against a different schema.
+        """
+        for use in self._parsed.find_all(exp.Use):
+            kind = use.args.get("kind")
+            # `USE WAREHOUSE ...` selects compute, not a namespace, and does
+            # not affect how table names resolve.
+            if kind and kind.name.upper() == "WAREHOUSE":
+                continue
+            return True
+        # `SET SCHEMA 'x'` / `SET CATALOG 'x'` rebind resolution through a
+        # structured setting rather than a search path.
+        rebinding_settings = {
+            "schema",
+            "current_schema",
+            "current schema",
+            "catalog",
+        }
+        if any(
+            key.strip('"').lower() in rebinding_settings for key in self.get_settings()
+        ):
+            return True
+        # A `set_config()` with a non-literal setting name may set
+        # `search_path` at runtime, so treat it as a schema change; literal
+        # names are handled by `changes_search_path`.
+        for func in self._parsed.find_all(exp.Anonymous):
+            if func.name.lower() == "set_config" and not (
+                func.expressions and isinstance(func.expressions[0], exp.Literal)
+            ):
+                return True
+        # `SET SCHEMA` / `SET CATALOG` forms that fall back to an opaque
+        # exp.Command: match the leading setting name, mirroring
+        # `changes_search_path`.
+        parsed = self._parsed
+        if isinstance(parsed, exp.Command) and parsed.name.upper() == "SET":
+            tokens = str(parsed.expression).replace("=", " ").split()
+            while tokens and tokens[0].upper() in {"SESSION", "LOCAL", "CURRENT"}:
+                tokens.pop(0)
+            if tokens and tokens[0].strip('"').strip("'").lower() in {
+                "schema",
+                "catalog",
+            }:
+                return True
+        return self.changes_search_path()
+
     def get_disallowed_tables(
         self,
         tables: set[str],
@@ -1282,6 +1409,15 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
         Modify the `LIMIT` or `TOP` value of the SQL statement inplace.
         """
         if method == LimitMethod.FORCE_LIMIT:
+            # `SHOW` statements (`SHOW TABLES`, `SHOW DATABASES`, `SHOW CREATE
+            # TABLE`, etc.) have no meaningful `LIMIT` slot to force. On
+            # MySQL/StarRocks, writing one renders a malformed statement with
+            # two `LIMIT` keywords that the engine rejects outright; on dialects
+            # like Snowflake it would render a valid `SHOW ... LIMIT`, but SHOW
+            # returns bounded metadata, so we skip it uniformly rather than
+            # special-case per dialect. Leave them untouched.
+            if isinstance(self._parsed, exp.Show):
+                return
             self._parsed.args["limit"] = exp.Limit(
                 expression=exp.Literal(this=str(limit), is_string=False)
             )
@@ -1408,7 +1544,30 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
             raise ValueError(f"Invalid RLS method: {method}")
 
         transformer = transformers[method](catalog, schema, predicates)
-        self._parsed = self._parsed.transform(transformer)
+
+        # Rewrite the real table reads -- the same set ``extract_tables_from_statement``
+        # authorizes -- so the filtered set equals the authorized set. (A CTE reference
+        # sharing a rule's table name is not a read here.)
+        seen: set[int] = set()
+        reads: list[exp.Table] = []
+        for scope in traverse_scope(self._parsed):
+            for source in scope.sources.values():
+                # dedupe by identity: a correlated LATERAL reaches one node twice
+                if (
+                    isinstance(source, exp.Table)
+                    and not is_cte(source, scope)
+                    and id(source) not in seen
+                ):
+                    seen.add(id(source))
+                    reads.append(source)
+
+        # Wrap the deepest reads first: a parenthesised-join head carries its join in
+        # its args, so wrapping an ancestor before its descendant would strand the
+        # descendant read's replacement off the live tree.
+        for node in sorted(reads, key=lambda read: read.depth, reverse=True):
+            replacement = transformer(node)
+            if replacement is not node:
+                node.replace(replacement)
 
 
 class KQLSplitState(enum.Enum):
@@ -1819,12 +1978,16 @@ class SQLScript:
     def has_unparseable_statement(self) -> bool:
         """
         True if any statement in the script cannot be fully modeled as an
-        AST whose table references Superset can enumerate. This covers two
-        cases that must both fail closed under strict scoping:
+        AST whose table references Superset can enumerate. This covers the
+        following cases, which must all fail closed under strict scoping:
 
         * SQLGlot ``exp.Command`` nodes: statements sqlglot recognises but
           cannot fully parse (e.g. dynamic SQL inside a stored-procedure
           call); ``extract_tables_from_statement`` cannot see the tables.
+        * ``exp.Show`` statements with no extractable target (e.g.
+          ``SHOW TABLES FROM some_schema``): the statement reads database
+          metadata, but there is no table reference for the per-table check
+          to enforce against.
         * Non-sqlglot engines (e.g. Kusto KQL): the statement class does
           not produce a sqlglot AST at all and its
           ``_extract_tables_from_statement`` returns an empty set, so the
@@ -1835,6 +1998,11 @@ class SQLScript:
                 return True
             if isinstance(statement._parsed, exp.Command):  # noqa: SLF001
                 return True
+            if (
+                isinstance(statement._parsed, exp.Show)  # noqa: SLF001
+                and not statement.tables
+            ):
+                return True
         return False
 
     def get_settings(self) -> dict[str, str | bool]:
@@ -1868,6 +2036,16 @@ class SQLScript:
         """
         return any(statement.is_destructive() for statement in self.statements)
 
+    def changes_default_schema(self) -> bool:
+        """
+        Check if any statement rebinds default schema resolution.
+
+        :return: True if any statement changes the schema (``USE``,
+            ``SET SCHEMA``) or the Postgres ``search_path`` used to resolve
+            unqualified table names
+        """
+        return any(statement.changes_default_schema() for statement in self.statements)
+
     def optimize(self) -> SQLScript:
         """
         Return optimized script.
@@ -1986,6 +2164,31 @@ def extract_tables_from_statement(
         except (ParseError, SupersetParseError):
             return set()
         sources = pseudo_query.find_all(exp.Table)
+    elif isinstance(statement, exp.Show):
+        # Structured metadata statements (`SHOW CREATE TABLE foo.bar`,
+        # `SHOW COLUMNS FROM foo`, ...) reference their target via dedicated
+        # args rather than query sources, so build the table references
+        # explicitly. Statements with no extractable target (e.g.
+        # `SHOW TABLES FROM some_schema`) yield an empty set and are treated
+        # as unparseable for authorization purposes (see
+        # `SQLScript.has_unparseable_statement`).
+        show_tables = {
+            Table(
+                source.name,
+                source.db if source.db != "" else None,
+                source.catalog if source.catalog != "" else None,
+            )
+            for source in statement.find_all(exp.Table)
+        }
+        if target := statement.args.get("target"):
+            db = statement.args.get("db")
+            show_tables.add(
+                Table(
+                    target.name if isinstance(target, exp.Expression) else str(target),
+                    db.name if isinstance(db, exp.Expression) else db,
+                )
+            )
+        return show_tables
     else:
         sources = [
             source
@@ -2006,41 +2209,22 @@ def extract_tables_from_statement(
 
 def is_cte(source: exp.Table, scope: Scope) -> bool:
     """
-    Is the source a CTE?
+    Does this reference resolve to a CTE rather than to a real table?
 
-    CTEs in the parent scope look like tables (and are represented by
-    exp.Table objects), but should not be considered as such;
-    otherwise a user with access to table `foo` could access any table
-    with a query like this:
-
-        WITH foo AS (SELECT * FROM target_table) SELECT * FROM foo
-
-    A CTE name is always a bare identifier: it can never carry a schema or
-    catalog qualifier. A schema/catalog-qualified reference therefore always
-    resolves to a physical table, even when its final name component happens to
-    match a CTE defined in scope. Such a reference must be reported as a real
-    table so it resolves to the correct object; otherwise
-    ``WITH orders AS (...) SELECT * FROM public.orders`` would treat the
-    qualified ``public.orders`` as the CTE and drop the physical table from the
-    extracted set.
-
-    Note: an unqualified reference is always resolved relative to the caller's
-    own schema/catalog before any downstream use, so treating a bare name that
-    matches a CTE as a CTE stays correct and is intentionally left unchanged
-    here.
+    A CTE reference is also an ``exp.Table``, so it must be excluded from a statement's
+    read tables, or a rule on a table could be evaded by wrapping it in a same-named
+    CTE. Resolve the name through ``Scope.cte_sources`` (not ``Scope.sources``, keyed by
+    ``alias_or_name``, which would hide a real table sharing a CTE's alias); a qualified
+    reference (schema or catalog) is always a table. Where sqlglot registers a name
+    differently than SQL scopes it (letter-case, a ``WITH RECURSIVE`` self/forward
+    reference), this errs toward reporting a table -- a spurious check, not a leak.
     """
     if source.db or source.catalog:
         # Qualified references are always physical tables, never CTEs.
         return False
 
-    parent_sources = scope.parent.sources if scope.parent else {}
-    ctes_in_scope = {
-        name
-        for name, parent_scope in parent_sources.items()
-        if isinstance(parent_scope, Scope) and parent_scope.scope_type == ScopeType.CTE
-    }
-
-    return source.name in ctes_in_scope
+    resolved = scope.cte_sources.get(source.name)
+    return isinstance(resolved, Scope) and resolved.scope_type == ScopeType.CTE
 
 
 T = TypeVar("T", str, None)
@@ -2071,6 +2255,17 @@ def remove_quotes(val: T) -> T:
     return val
 
 
+# Jinja macros that execute statements against the analytical database when
+# rendered; their table references are extracted before rendering, and the
+# macros are stubbed out during a validation-time render.
+PARTITION_MACRO_NAMES = (
+    "first_latest_partition",
+    "latest_partition",
+    "latest_partitions",
+    "latest_sub_partition",
+)
+
+
 def process_jinja_sql(
     sql: str, database: Database, template_params: Optional[dict[str, Any]] = None
 ) -> JinjaSQLResult:
@@ -2091,10 +2286,13 @@ def process_jinja_sql(
     :returns: JinjaSQLResult containing the processed script and table references
     :raises SupersetSecurityException: If SQLGlot is unable to parse the SQL statement
     :raises jinja2.exceptions.TemplateError: If the Jinjafied SQL could not be rendered
+    :raises SupersetParseError: If a partition macro references a table that
+        cannot be determined statically
     """
 
     from superset.jinja_context import (  # pylint: disable=import-outside-toplevel
         get_template_processor,
+        NoOpTemplateProcessor,
     )
 
     processor = get_template_processor(database)
@@ -2102,37 +2300,74 @@ def process_jinja_sql(
 
     tables = set()
 
+    def raise_for_unresolvable_macro() -> Any:
+        raise SupersetParseError(
+            sql,
+            database.db_engine_spec.engine,
+            message=(
+                "Unable to determine the table referenced by a partition "
+                "macro; use a single constant table reference"
+            ),
+        )
+
     for node in ast.find_all(nodes.Call):
-        if isinstance(node.node, nodes.Getattr) and node.node.attr in (
-            "latest_partition",
-            "latest_sub_partition",
+        if (
+            isinstance(node.node, nodes.Getattr)
+            and node.node.attr in PARTITION_MACRO_NAMES
         ):
-            # Try to extract the table referenced in the macro.
+            # Extract the table referenced in the macro. The reference must
+            # be statically evaluable; otherwise raise rather than render.
             try:
+                if len(node.args) != 1:
+                    raise nodes.Impossible()
                 tables.add(
                     Table(
                         *[
                             remove_quotes(part.strip())
                             for part in node.args[0].as_const().split(".")[::-1]
-                            if len(node.args) == 1
                         ]
                     )
                 )
             except nodes.Impossible:
-                pass
+                raise_for_unresolvable_macro()
 
             # Replace the potentially problematic Jinja macro with some benign SQL.
             node.__class__ = nodes.TemplateData
             node.fields = nodes.TemplateData.fields
             node.data = "NULL"
 
-    # re-render template back into a string
-    code = processor.env.compile(ast)
-    template = Template.from_code(processor.env, code, globals=processor.env.globals)
-    rendered_sql = template.render(processor.get_context(), **(template_params or {}))
+    # Render the neutralized template once, using the same context
+    # ``process_template`` builds at execution time, so the validated SQL
+    # matches the executed SQL. A no-op processor runs the raw SQL at
+    # execution time, so validate that raw SQL directly.
+    if isinstance(processor, NoOpTemplateProcessor):
+        rendered_sql = processor.process_template(sql)
+    else:
+        code = processor.env.compile(ast)
+        template = Template.from_code(
+            processor.env,
+            code,
+            globals=processor.env.globals,
+        )
+        # Replace live partition macros with stubs so a call that survives
+        # neutralization (e.g. via a dynamic attribute lookup) does not
+        # execute during this render.
+        context = processor.get_template_context(**(template_params or {}))
+        if (engine := getattr(processor, "engine", None)) and isinstance(
+            context.get(engine), dict
+        ):
+            context[engine] = {
+                key: (
+                    (lambda *args, **kwargs: raise_for_unresolvable_macro())
+                    if key in PARTITION_MACRO_NAMES
+                    else value
+                )
+                for key, value in context[engine].items()
+            }
+        rendered_sql = template.render(context)
 
     parsed_script = SQLScript(
-        processor.process_template(rendered_sql),
+        rendered_sql,
         engine=database.db_engine_spec.engine,
     )
     for parsed_statement in parsed_script.statements:

superset/sqllab/api.py

@@ -258,6 +258,14 @@ class SqlLabRestApi(BaseSupersetApi):
                                 else template_params
                             )
                             if template_params:
+                                # Check access before rendering the Jinja
+                                # template (mirrors the SQL Lab execute path).
+                                security_manager.raise_for_access(
+                                    database=database,
+                                    sql=sql,
+                                    template_params=template_params,
+                                    force_dataset_match=True,
+                                )
                                 template_processor = get_template_processor(
                                     database=database
                                 )

superset/sqllab/sqllab_execution_context.py

@@ -22,9 +22,12 @@ from dataclasses import dataclass
 from typing import Any, cast, TYPE_CHECKING
 
 from flask import g
+from flask_babel import gettext as __
 from sqlalchemy.orm.exc import DetachedInstanceError
 
 from superset import is_feature_enabled
+from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
+from superset.exceptions import SupersetErrorException
 from superset.models.sql_lab import Query
 from superset.sql.parse import CTASMethod
 from superset.utils import core as utils, json
@@ -128,9 +131,45 @@ class SqlJsonExecutionContext:  # pylint: disable=too-many-instance-attributes
         if self.catalog is None:
             self.catalog = database.get_default_catalog()
         if self.select_as_cta:
+            self._validate_ctas_is_allowed(database)
             schema_name = self._get_ctas_target_schema_name(database)
             self.create_table_as_select.target_schema_name = schema_name  # type: ignore
 
+    def _validate_ctas_is_allowed(self, database: Database) -> None:
+        """
+        Enforce the per-database CTAS/CVAS grants server-side.
+
+        The database's ``allow_ctas``/``allow_cvas`` flags are checked at
+        submission, mirroring the ``allow_dml`` gate on the execution path.
+        """
+        ctas = cast(CreateTableAsSelect, self.create_table_as_select)
+        if ctas.ctas_method == CTASMethod.TABLE and not database.allow_ctas:
+            raise SupersetErrorException(
+                SupersetError(
+                    message=__(
+                        "This database does not allow creating tables from "
+                        "queries (CTAS). Please contact your administrator "
+                        "for more assistance."
+                    ),
+                    error_type=SupersetErrorType.QUERY_SECURITY_ACCESS_ERROR,
+                    level=ErrorLevel.ERROR,
+                ),
+                status=403,
+            )
+        if ctas.ctas_method == CTASMethod.VIEW and not database.allow_cvas:
+            raise SupersetErrorException(
+                SupersetError(
+                    message=__(
+                        "This database does not allow creating views from "
+                        "queries (CVAS). Please contact your administrator "
+                        "for more assistance."
+                    ),
+                    error_type=SupersetErrorType.QUERY_SECURITY_ACCESS_ERROR,
+                    level=ErrorLevel.ERROR,
+                ),
+                status=403,
+            )
+
     def _get_ctas_target_schema_name(self, database: Database) -> str | None:
         if database.force_ctas_schema:
             return database.force_ctas_schema

superset/tasks/context.py

@@ -20,9 +20,10 @@ import logging
 import threading
 import time
 import traceback
+from contextlib import nullcontext
 from typing import Any, Callable, cast, TYPE_CHECKING, TypeVar
 
-from flask import current_app
+from flask import current_app, has_app_context
 from superset_core.tasks.types import (
     TaskContext as CoreTaskContext,
     TaskProperties,
@@ -256,7 +257,7 @@ class TaskContext(CoreTaskContext):
 
             if self._has_pending_updates:
                 # Need app context for DB operations in timer thread
-                if self._app:
+                if self._app and not has_app_context():
                     with self._app.app_context():
                         self._write_to_db()
                 else:
@@ -475,7 +476,8 @@ class TaskContext(CoreTaskContext):
             )
 
         if self._app:
-            with self._app.app_context():
+            ctx = self._app.app_context() if not has_app_context() else nullcontext()
+            with ctx:
                 # Check if task already has an error (preserve original context)
                 task = self._task
                 original_error = task.properties_dict.get("error_message")
@@ -550,7 +552,8 @@ class TaskContext(CoreTaskContext):
                 )
                 return
 
-            with self._app.app_context():
+            ctx = self._app.app_context() if not has_app_context() else nullcontext()
+            with ctx:
                 from superset.commands.tasks.update import UpdateTaskCommand
 
                 task = self._task
@@ -637,7 +640,8 @@ class TaskContext(CoreTaskContext):
         # If aborting/aborted but handlers haven't run yet, run them now
         # (This catches the case where task ended before listener detected abort)
         if self._app:
-            with self._app.app_context():
+            ctx = self._app.app_context() if not has_app_context() else nullcontext()
+            with ctx:
                 task = self._task
                 if task.status in ABORT_STATES and not self._abort_detected:
                     self._trigger_abort_handlers()

superset/tasks/manager.py

@@ -25,6 +25,7 @@ from typing import Any, Callable, TYPE_CHECKING
 from uuid import UUID
 
 import redis
+from flask import has_app_context
 from superset_core.tasks.types import TaskProperties, TaskScope
 
 from superset.async_events.cache_backend import (
@@ -258,7 +259,7 @@ class TaskManager:
             return remaining if remaining > 0 else 0
 
         def get_task() -> "Task | None":
-            if app:
+            if app and not has_app_context():
                 with app.app_context():
                     return TaskDAO.find_one_or_none(uuid=task_uuid)
             return TaskDAO.find_one_or_none(uuid=task_uuid)
@@ -461,7 +462,7 @@ class TaskManager:
         :param callback: Function to invoke
         :param app: Flask app for context, or None
         """
-        if app:
+        if app and not has_app_context():
             with app.app_context():
                 callback()
         else:
@@ -659,7 +660,7 @@ class TaskManager:
 
         def check_database() -> bool:
             # Need app context for database access
-            if app:
+            if app and not has_app_context():
                 with app.app_context():
                     return cls._check_abort_status(task_uuid)
             else:

superset/tasks/slack.py

@@ -18,17 +18,27 @@ import logging
 
 from flask import current_app
 
+from superset.constants import CACHE_DISABLED_TIMEOUT
 from superset.extensions import celery_app
+from superset.utils.decorators import transaction
 from superset.utils.slack import get_channels
 
 logger = logging.getLogger(__name__)
 
 
 @celery_app.task(name="slack.cache_channels")
+@transaction()
 def cache_channels() -> None:
     cache_timeout = current_app.config["SLACK_CACHE_TIMEOUT"]
     retry_count = current_app.config.get("SLACK_API_RATE_LIMIT_RETRY_COUNT", 2)
 
+    if cache_timeout == CACHE_DISABLED_TIMEOUT:
+        logger.warning(
+            "Skipping Slack channels cache warm-up because "
+            "SLACK_CACHE_TIMEOUT disables caching"
+        )
+        return
+
     logger.info(
         "Starting Slack channels cache warm-up task "
         "(cache_timeout=%ds, retry_count=%d)",
@@ -37,7 +47,11 @@ def cache_channels() -> None:
     )
 
     try:
-        get_channels(force=True, cache_timeout=cache_timeout)
+        get_channels(
+            force=True,
+            cache_timeout=cache_timeout,
+            raise_on_cache_write_error=True,
+        )
     except Exception as ex:
         logger.exception(
             "Failed to cache Slack channels: %s. "

superset/utils/core.py

@@ -816,6 +816,7 @@ def pessimistic_connection_handling(some_engine: Engine) -> None:
             # the SELECT of a scalar value without a table is
             # appropriately formatted for the backend
             connection.scalar(select(1))
+            connection.rollback()  # pylint: disable=consider-using-transaction
         except exc.DBAPIError as err:
             # catch SQLAlchemy's DBAPIError, which is a wrapper
             # for the DBAPI's exception.  It includes a .connection_invalidated
@@ -828,6 +829,7 @@ def pessimistic_connection_handling(some_engine: Engine) -> None:
                 # here also causes the whole connection pool to be invalidated
                 # so that all stale connections are discarded.
                 connection.scalar(select(1))
+                connection.rollback()  # pylint: disable=consider-using-transaction
             else:
                 raise
         finally:

superset/utils/decorators.py

@@ -36,7 +36,21 @@ if TYPE_CHECKING:
     from superset.stats_logger import BaseStatsLogger
 
 
-def statsd_gauge(metric_prefix: str | None = None) -> Callable[..., Any]:
+def record_statsd_gauge_failure(metric_prefix: str, ex: Exception) -> None:
+    """Record a warning or error gauge using the shared exception contract."""
+    try:
+        status = getattr(ex, "status", None)
+    except Exception:  # pylint: disable=broad-exception-caught
+        status = None
+    suffix = "warning" if isinstance(status, int) and status < 500 else "error"
+    app.config["STATS_LOGGER"].gauge(f"{metric_prefix}.{suffix}", 1)
+
+
+def statsd_gauge(
+    metric_prefix: str | None = None,
+    *,
+    ignored_exceptions: tuple[type[Exception], ...] = (),
+) -> Callable[..., Any]:
     def decorate(f: Callable[..., Any]) -> Callable[..., Any]:
         """
         Handle sending statsd gauge metric from any method or function
@@ -48,13 +62,10 @@ def statsd_gauge(metric_prefix: str | None = None) -> Callable[..., Any]:
                 result = f(*args, **kwargs)
                 app.config["STATS_LOGGER"].gauge(f"{metric_prefix_}.ok", 1)
                 return result
+            except ignored_exceptions:
+                raise
             except Exception as ex:
-                if (
-                    hasattr(ex, "status") and ex.status < 500  # pylint: disable=no-member
-                ):
-                    app.config["STATS_LOGGER"].gauge(f"{metric_prefix_}.warning", 1)
-                else:
-                    app.config["STATS_LOGGER"].gauge(f"{metric_prefix_}.error", 1)
+                record_statsd_gauge_failure(metric_prefix_, ex)
                 raise
 
         return wrapped

superset/utils/number_format.py

@@ -0,0 +1,607 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Server-side port of the d3-format based number and currency formatters used by
+the Table and Pivot Table chart plugins.
+
+Report notifications that embed a chart as text build the table in Python and
+have no access to the frontend formatters, so chart number/currency format
+configuration has to be reproduced here to render the same values an end user
+sees in the browser.
+
+Only d3-format specifiers (and the ``SMART_NUMBER`` pseudo-formats) are ported.
+The duration, memory, and length formatters depend on separate frontend
+factories and are explicitly rejected, causing the public wrapper to preserve
+the raw value rather than silently misformat it. The fill/align/zero/width d3
+flags are likewise rejected because report text has no equivalent of the
+frontend's padding behavior. Accounting-parenthesis and space-sign modes are
+supported.
+"""
+
+from __future__ import annotations
+
+import math
+import re
+from decimal import Decimal, ROUND_HALF_UP
+from functools import lru_cache
+from typing import Any, Iterable
+
+from babel.numbers import format_currency, get_currency_symbol
+from flask import current_app
+from flask_babel import get_locale
+
+SMART_NUMBER: str = "SMART_NUMBER"
+SMART_NUMBER_SIGNED: str = "SMART_NUMBER_SIGNED"
+AUTO_CURRENCY: str = "AUTO"
+
+UNSUPPORTED_FRONTEND_PRESETS: frozenset[str] = frozenset(
+    {
+        "DURATION",
+        "DURATION_SUB",
+        "DURATION_COL",
+        "MEMORY_DECIMAL",
+        "MEMORY_BINARY",
+        "MEMORY_TRANSFER_RATE_DECIMAL",
+        "MEMORY_TRANSFER_RATE_BINARY",
+        "LENGTH",
+        "LENGTH_CM_KM",
+        "LENGTH_CM_M",
+    }
+)
+
+DEFAULT_LOCALE: str = "en"
+CURRENCY_SYMBOL_LOCALE: str = "en_US"
+
+# SI prefixes keyed by their power-of-1000 exponent, mirroring d3-format.
+SI_PREFIXES: dict[int, str] = {
+    -8: "y",
+    -7: "z",
+    -6: "a",
+    -5: "f",
+    -4: "p",
+    -3: "n",
+    -2: "µ",
+    -1: "m",
+    0: "",
+    1: "k",
+    2: "M",
+    3: "G",
+    4: "T",
+    5: "P",
+    6: "E",
+    7: "Z",
+    8: "Y",
+}
+
+# d3-format specifier grammar:
+# [[fill]align][sign][symbol][0][width][,][.precision][~][type]
+D3_FORMAT_RE: re.Pattern[str] = re.compile(
+    r"^(?:(.)?([<>=^]))?([+\-( ])?([$#])?(0)?(\d+)?(,)?(?:\.(\d+))?(~)?([a-z%])?$",
+    re.IGNORECASE,
+)
+
+
+def resolve_auto_currency(
+    currency: dict[str, Any],
+    detected_currency: str | None,
+    currency_context: Iterable[Any] | float | None = None,
+    fallback_to_detected: bool = True,
+) -> dict[str, Any]:
+    """
+    Resolve an ``AUTO`` currency to the code detected from the data.
+
+    Mirrors ``currency-format/utils.ts::resolveAutoCurrency`` and the per-cell
+    handling in the Table and Pivot Table plugins. A single valid currency in
+    ``currency_context`` takes precedence over the query-wide detection. Mixed
+    cell currencies deliberately keep ``AUTO`` so the caller renders a neutral
+    number. Empty cell context can use the detected fallback when the plugin's
+    behavior allows it.
+
+    :param currency_context: the currencies contributing to a cell. A dense
+        pivot cell provides an iterable of codes, but a sparse 2D pivot passes a
+        scalar ``NaN`` float for a missing cross-product cell (hence the ``float``
+        arm); a missing/NaN/non-iterable context is treated as empty.
+    :return: a copied config containing the detected code, or the input config
+    """
+    if currency.get("symbol") != AUTO_CURRENCY:
+        return currency
+
+    if currency_context is not None:
+        # A dense pivot cell carries an iterable of currency codes, but a sparse
+        # 2D pivot leaves missing cross-product cells as a scalar missing value
+        # (``np.nan``; pandas never runs the union aggregator for them). Test
+        # positively for an iterable so any non-iterable sentinel (``np.nan``,
+        # ``pd.NA``, ``pd.NaT``) falls to the empty-context path instead of
+        # raising and taking down the whole report.
+        context_values: list[Any] = (
+            list(currency_context) if isinstance(currency_context, Iterable) else []
+        )
+        normalized_currencies = {
+            normalized
+            for value in context_values
+            if (normalized := normalize_currency(value)) is not None
+        }
+        if len(normalized_currencies) > 1:
+            return currency
+        if context_values and (cell_currency := normalize_currency(context_values[0])):
+            return {**currency, "symbol": cell_currency}
+        if not fallback_to_detected:
+            return currency
+
+    if detected_currency := normalize_currency(detected_currency):
+        return {**currency, "symbol": detected_currency}
+    return currency
+
+
+def normalize_currency(value: Any) -> str | None:
+    """
+    Normalize a possible ISO-4217 code for AUTO currency resolution.
+
+    Mirrors ``currency-format/CurrencyFormatter.ts::normalizeCurrency``:
+    non-strings and values other than three ASCII letters are rejected, while
+    valid strings are stripped and upper-cased.
+
+    :return: the normalized three-letter code, or ``None``
+    """
+    if not isinstance(value, str):
+        return None
+    normalized = value.strip().upper()
+    return normalized if re.fullmatch(r"[A-Z]{3}", normalized) else None
+
+
+def format_number_with_config(
+    d3_format: str | None,
+    currency: dict[str, Any] | None,
+    value: Any,
+) -> Any:
+    """
+    Format ``value`` using a d3-format string and optional currency config.
+
+    This is the report-side entry point corresponding to
+    ``currency-format/CurrencyFormatter.ts::format`` and the formatter invoked
+    by the Table and Pivot Table plugins.
+
+    :param d3_format: a d3-format specifier (e.g. ``",.2f"``) or ``SMART_NUMBER``
+    :param currency: ``{"symbol": <ISO 4217>, "symbolPosition": "prefix"|"suffix"}``
+    :param value: the raw value to format
+    :return: the formatted string, or the value unchanged when it is not a
+        number that can be formatted
+    """
+    if value is None:
+        return ""
+    if isinstance(value, bool) or not isinstance(value, (int, float, Decimal)):
+        return value
+    if isinstance(value, Decimal):
+        value = float(value)
+    if math.isnan(value) or math.isinf(value):
+        return ""
+
+    try:
+        if currency and currency.get("symbol"):
+            # the frontend strips the currency symbol from the d3 format and
+            # falls back to SMART_NUMBER when no explicit format is set
+            number_format = (d3_format or SMART_NUMBER).replace("$", "")
+            formatted = format_numeric(number_format, value)
+            if currency["symbol"] == AUTO_CURRENCY:
+                return formatted
+            try:
+                return apply_currency(formatted, currency)
+            except Exception:  # pylint: disable=broad-except  # noqa: BLE001
+                return formatted
+        if not d3_format:
+            return raw_string(value)
+        return format_numeric(d3_format, value)
+    except Exception:  # pylint: disable=broad-except  # noqa: BLE001
+        # never let an unexpected value break a whole report table
+        return raw_string(value)
+
+
+def format_numeric(d3_format: str, value: float) -> str:
+    """
+    Format ``value`` according to a d3 number format.
+
+    Delegates to the port of ``createSmartNumberFormatter.ts`` for the two smart
+    pseudo-formats and to the port of ``d3-format/src/locale.js`` for d3
+    specifiers. Registered frontend-only factories are rejected explicitly.
+
+    :return: a formatted number string
+    """
+    if d3_format in UNSUPPORTED_FRONTEND_PRESETS:
+        raise ValueError(f"Frontend preset {d3_format!r} is not available in reports")
+    if d3_format in (SMART_NUMBER, SMART_NUMBER_SIGNED):
+        return format_smart_number(value, signed=d3_format == SMART_NUMBER_SIGNED)
+    return format_d3(d3_format, value)
+
+
+def format_d3(d3_format: str, value: float) -> str:
+    """
+    Format ``value`` with a d3-format specifier.
+
+    Mirrors ``d3-format/src/locale.js`` and ``formatTypes.js``. Supports the
+    subset of the specifier grammar the Table/Pivot plugins emit:
+    the ``+ - ( space`` sign modes, the ``$`` currency prefix, the ``,`` group
+    separator, ``.precision``, the ``~`` trim flag, and the ``s`` (SI), ``r``
+    (significant), ``d`` (integer), ``f``/``e``/``g``/``%`` numeric types.
+    Returns the formatted string and raises ``ValueError`` for an unparseable
+    specifier. Padding flags are rejected because they cannot be represented by
+    the report table path.
+
+    :return: a d3-compatible formatted string
+    """
+    match = D3_FORMAT_RE.match(d3_format)
+    if not match:
+        raise ValueError(d3_format)
+    if any(match.group(index) for index in (1, 2, 5, 6)):
+        raise ValueError(f"d3 padding is not supported in reports: {d3_format!r}")
+
+    sign_mode = match.group(3) or "-"
+    currency_symbol = match.group(4) == "$"
+    comma = "," if match.group(7) else ""
+    precision = int(match.group(8)) if match.group(8) is not None else None
+    trim = bool(match.group(9))
+    type_ = (match.group(10) or "").lower()
+
+    if type_ == "n":
+        comma = ","
+        type_ = "g"
+
+    formatted = format_d3_magnitude(
+        type_, abs(value), precision, trim, comma, d3_format
+    )
+
+    if currency_symbol:
+        formatted = f"${formatted}"
+    return apply_sign(formatted, value, sign_mode)
+
+
+def format_d3_magnitude(
+    type_: str,
+    magnitude: float,
+    precision: int | None,
+    trim: bool,
+    comma: str,
+    d3_format: str,
+) -> str:
+    """
+    Render the unsigned numeric portion of a parsed d3 specifier.
+
+    Mirrors the formatter dispatch in ``d3-format/src/locale.js`` and
+    ``formatTypes.js``. The result excludes sign and currency decoration.
+    """
+    if type_ == "s":
+        return format_si(
+            magnitude, max(1, precision if precision is not None else 6), trim
+        )
+    if type_ == "r":
+        return format_significant(
+            magnitude,
+            max(1, precision if precision is not None else 6),
+            trim,
+            comma,
+        )
+    if type_ == "":
+        return format_general(
+            magnitude, precision if precision is not None else 12, True, comma
+        )
+    if type_ == "d":
+        formatted = format(int(quantize_half_up(magnitude, 0)), f"{comma}d")
+    elif type_ in ("f", "%"):
+        precision = precision if precision is not None else 6
+        scaled = magnitude * 100 if type_ == "%" else magnitude
+        suffix = "%" if type_ == "%" else ""
+        if scaled >= 1e21:
+            formatted = normalize_exponent(repr(float(scaled))) + suffix
+        else:
+            rounded = quantize_half_up(scaled, precision)
+            formatted = format(rounded, f"{comma}.{precision}f") + suffix
+    elif type_ == "e":
+        formatted = format_exponential(
+            magnitude, precision if precision is not None else 6
+        )
+    elif type_ == "g":
+        formatted = format_general(
+            magnitude, precision if precision is not None else 6, trim, comma
+        )
+    else:
+        raise ValueError(d3_format)
+    return trim_trailing_zeros(formatted) if trim else formatted
+
+
+def apply_sign(formatted: str, value: float, sign_mode: str) -> str:
+    """
+    Decorate a formatted magnitude with the d3 sign mode.
+
+    Negative values get a leading ``-`` (or wrapping parentheses for the ``(``
+    accounting mode); positive values get a ``+`` or a leading space only for the
+    ``+`` and space modes respectively. Mirrors the sign decoration in
+    ``d3-format/src/locale.js``.
+
+    :return: the signed or accounting-decorated string
+    """
+    if value < 0:
+        return f"({formatted})" if sign_mode == "(" else f"-{formatted}"
+    if sign_mode == "+":
+        return f"+{formatted}"
+    if sign_mode == " ":
+        return f" {formatted}"
+    return formatted
+
+
+def format_default(value: float, comma: str) -> str:
+    """
+    Format ``value`` the way d3's default (no-type) specifier does.
+
+    d3 aliases an omitted type to ``.12~g``. This preserves fixed notation from
+    ``1e-6`` through twelve significant integer digits, then uses exponent
+    notation outside that range. Mirrors the omitted-type alias in
+    ``d3-format/src/formatSpecifier.js``.
+
+    :return: the ``.12~g`` representation
+    """
+    return format_general(value, 12, True, comma)
+
+
+def format_general(value: float, precision: int, trim: bool, comma: str = "") -> str:
+    """
+    Format d3's ``g`` type with JavaScript ``toPrecision`` thresholds.
+
+    Mirrors ``d3-format/src/formatTypes.js`` and returns fixed or exponential
+    notation with the requested significant-digit precision.
+    """
+    precision = max(1, precision)
+    rounded = round_to_significant(value, precision)
+    exponent = decimal_exponent(rounded)
+    if value and (exponent < -6 or exponent >= precision):
+        formatted = format_exponential(value, precision - 1)
+    else:
+        formatted = format_significant(value, precision, False, comma)
+    return trim_trailing_zeros(formatted) if trim else formatted
+
+
+def format_exponential(value: float, precision: int) -> str:
+    """
+    Format d3's ``e`` type using binary-float, half-up rounding.
+
+    Mirrors the ``e`` formatter in ``d3-format/src/formatTypes.js`` and returns
+    an exponent without redundant leading zeros.
+    """
+    rounded = round_to_significant(value, precision + 1)
+    exponent = decimal_exponent(rounded)
+    mantissa = rounded / (10**exponent) if rounded else 0.0
+    return f"{mantissa:.{precision}f}e{exponent:+d}"
+
+
+def format_smart_number(value: float, signed: bool = False) -> str:
+    """
+    Format ``value`` the way the frontend ``SMART_NUMBER`` formatter does.
+
+    The notation is chosen by magnitude: SI prefixes (with ``G`` shown as ``B``)
+    for ``abs(value) >= 1000``, two decimals down to ``1``, four decimals down to
+    ``0.001``, a micro (``µ``) suffix down to ``1e-6``, and SI prefixes again
+    below that. When ``signed`` is set, positive values are prefixed with ``+``.
+    Mirrors ``number-format/factories/createSmartNumberFormatter.ts``.
+
+    :return: the adaptive frontend-compatible number string
+    """
+    if value == 0:
+        body = "0"
+    else:
+        absolute = abs(value)
+        if absolute >= 1000:
+            body = format_si(value, 3, trim=True, billions=True)
+        elif absolute >= 1:
+            body = trim_trailing_zeros(format(quantize_half_up(value, 2), ".2f"))
+        elif absolute >= 0.001:
+            body = trim_trailing_zeros(format(quantize_half_up(value, 4), ".4f"))
+        elif absolute > 0.000001:
+            body = format_si(value * 1000000, 3, trim=True) + "µ"
+        else:
+            body = format_si(value, 3, trim=True)
+    prefix = "+" if signed and value > 0 else ""
+    return prefix + body
+
+
+def format_si(value: float, precision: int, trim: bool, billions: bool = False) -> str:
+    """
+    Format ``value`` with an SI prefix to ``precision`` significant digits.
+
+    Rounds to ``precision`` significant figures first, then scales into the
+    nearest power-of-1000 bracket (clamped to the ``y``..``Y`` range) and appends
+    the matching SI symbol. Rounding before the divide matches d3 and keeps
+    ``4725`` at ``4.73k`` (the inexact ``4.725`` mantissa would round to
+    ``4.72k``), and lets a value that rounds up into the next bracket pick the
+    right symbol (``999.5k`` -> ``1M``). With ``billions`` set, the ``G`` (giga)
+    symbol is rendered as ``B``. Mirrors d3's
+    ``formatPrefixAuto.js``/``formatRounded.js`` combination.
+
+    :return: a significant-digit mantissa followed by its SI prefix
+    """
+    if value == 0:
+        return format_significant(0.0, precision, trim)
+
+    rounded = round_to_significant(value, precision)
+    exponent = max(-8, min(8, math.floor(math.log10(abs(rounded))) // 3))
+    mantissa = rounded / (10 ** (exponent * 3))
+
+    symbol = SI_PREFIXES[exponent]
+    if billions and symbol == "G":
+        symbol = "B"
+
+    return format_significant(mantissa, precision, trim) + symbol
+
+
+def format_significant(
+    value: float, precision: int, trim: bool, comma: str = ""
+) -> str:
+    """
+    Format to `precision` significant digits in fixed-point notation.
+
+    Serves both the d3 `r` type and SI mantissas, and avoids the scientific
+    notation Python's `g` would switch to. Mirrors the fixed representation
+    produced by ``d3-format/src/formatRounded.js``.
+
+    :return: a fixed-point significant-digit string
+    """
+    rounded = round_to_significant(value, precision)
+    decimals = decimals_for_significant(rounded, precision)
+    formatted = format(rounded, f"{comma}.{decimals}f")
+    return trim_trailing_zeros(formatted) if trim else formatted
+
+
+def round_to_significant(value: float, precision: int) -> float:
+    """
+    Round ``value`` to ``precision`` significant digits.
+
+    The number of decimal places to keep is derived from the value's order of
+    magnitude (``precision - 1 - floor(log10(abs(value)))``) and the rounding is
+    half away from zero, matching d3-format's ``formatDecimalParts`` path.
+
+    :return: the rounded binary-float value
+    """
+    if value == 0:
+        return 0.0
+    return float(quantize_half_up(value, precision - 1 - decimal_exponent(value)))
+
+
+def quantize_half_up(value: float, decimals: int) -> Decimal:
+    """
+    Round to `decimals` places, half away from zero, matching d3-format.
+
+    Quantizes the binary float value (not its decimal string) so the result
+    matches d3, which rounds the IEEE-754 value: ``2.675`` is ``2.67`` because it
+    is really ``2.67499...``, while an exact ``0.125`` rounds up to ``0.13``.
+    This supplies the rounding semantics of ``d3-format/src/formatTypes.js``.
+
+    :return: a ``Decimal`` rounded at the requested decimal place
+    """
+    return Decimal(value).quantize(Decimal(1).scaleb(-decimals), rounding=ROUND_HALF_UP)
+
+
+def decimals_for_significant(value: float, precision: int) -> int:
+    """
+    Return fixed-point decimal places needed for significant-digit formatting.
+
+    This is the report-side equivalent of the exponent adjustment in
+    ``d3-format/src/formatRounded.js``.
+    """
+    integer_digits = 1 if value == 0 else decimal_exponent(value) + 1
+    return max(0, precision - integer_digits)
+
+
+def decimal_exponent(value: float) -> int:
+    """
+    Return the base-10 exponent without ``log10`` boundary drift.
+
+    Used where d3-format derives an exponent through ``formatDecimalParts``.
+    """
+    return Decimal(repr(value)).adjusted() if value else 0
+
+
+def normalize_exponent(formatted: str) -> str:
+    """
+    Drop exponent leading zeros (``1e+07`` to ``1e+7``), as d3 does.
+
+    :return: the exponent string style emitted by ``d3-format``
+    """
+    return re.sub(r"([eE][+-])0*(\d)", r"\1\2", formatted)
+
+
+def get_currency_locale() -> str:
+    """
+    Return the request locale, or the configured default outside a request.
+
+    Report tasks run with a Flask application context but without a request, so
+    Flask-Babel can return ``None``. The config fallback keeps Celery-rendered
+    reports aligned with the locale supplied to the frontend at bootstrap. The
+    result feeds the locale argument used by ``currency-format/symbolPosition.ts``.
+
+    :return: a Babel locale identifier, always with a safe default
+    """
+    try:
+        if locale := get_locale():
+            return str(locale)
+    except RuntimeError:
+        pass
+
+    try:
+        return str(current_app.config.get("BABEL_DEFAULT_LOCALE") or DEFAULT_LOCALE)
+    except RuntimeError:
+        return DEFAULT_LOCALE
+
+
+@lru_cache(maxsize=None)
+def resolve_symbol_position(code: str, locale: str) -> str:
+    """
+    Derive the symbol position from the locale's convention for the currency.
+
+    Mirrors ``currency-format/symbolPosition.ts::resolveSymbolPosition`` and
+    returns ``"prefix"`` on invalid locale/currency input.
+    """
+    try:
+        sample = format_currency(1, code, locale=locale)
+        first_digit = next(i for i, char in enumerate(sample) if char.isdigit())
+        return "prefix" if first_digit > 0 else "suffix"
+    except Exception:  # pylint: disable=broad-except  # noqa: BLE001
+        return "prefix"
+
+
+def apply_currency(formatted: str, currency: dict[str, Any]) -> str:
+    """
+    Add a localized currency symbol to an already formatted number.
+
+    Mirrors ``currency-format/CurrencyFormatter.ts::format``: percentage signs
+    are removed, explicit positions win, and an unset position is locale-driven.
+
+    :return: the number with a prefix or suffix currency symbol
+    """
+    normalized = formatted.replace("%", "")
+    code = currency["symbol"]
+    symbol = get_currency_symbol(code, locale=CURRENCY_SYMBOL_LOCALE) or code
+    position = currency.get("symbolPosition")
+    if position not in ("prefix", "suffix"):
+        position = resolve_symbol_position(code, get_currency_locale())
+    if position == "prefix":
+        return f"{symbol} {normalized}"
+    return f"{normalized} {symbol}"
+
+
+def trim_trailing_zeros(formatted: str) -> str:
+    """
+    Remove insignificant fractional zeros while preserving suffixes.
+
+    Mirrors ``d3-format/src/formatTrim.js`` for decimal, exponent, and percent
+    strings and returns the compact representation.
+    """
+    suffix = "%" if formatted.endswith("%") else ""
+    body = formatted[: -len(suffix)] if suffix else formatted
+    coefficient, separator, exponent = body.partition("e")
+    if "." in coefficient:
+        coefficient = coefficient.rstrip("0").rstrip(".")
+    exponent_suffix = f"{separator}{exponent}" if separator else ""
+    return coefficient + exponent_suffix + suffix
+
+
+def raw_string(value: float) -> str:
+    """
+    Convert an unformatted number to the frontend-like neutral representation.
+
+    Integral floats lose their Python-only ``.0`` suffix. The result is the
+    safe fallback used by ``CurrencyFormatter.ts`` and invalid format handling.
+    """
+    if isinstance(value, float) and value.is_integer():
+        return str(int(value))
+    return str(value)

superset/utils/rls.py

@@ -17,18 +17,41 @@
 
 from __future__ import annotations
 
+import hashlib
 from typing import Any, TYPE_CHECKING
 
 from sqlalchemy import and_, or_
 
-from superset import db
+from superset import db, security_manager
 from superset.sql.parse import Table
+from superset.utils import json
+from superset.utils.core import get_user_id
 
 if TYPE_CHECKING:
     from superset.models.core import Database
     from superset.sql.parse import BaseSQLStatement
 
 
+def _get_cache_identity() -> str:
+    """
+    Build a stable per-session identity to key the parse-failure sentinel on.
+
+    Logged-in users have a stable numeric id from ``get_user_id()``. Guest
+    users (embedded) don't -- ``get_user_id()`` always returns ``None`` for
+    them -- so different guest tokens with different RLS scopes would
+    otherwise all collapse onto the same "user-None" sentinel and share cache
+    entries. Key those on a hash of the guest token's own RLS rules instead,
+    so distinct guest scopes stay isolated from one another.
+    """
+    if guest_user := security_manager.get_current_guest_user_if_guest():
+        rls_rules = guest_user.guest_token.get("rls_rules", [])
+        digest = hashlib.sha256(
+            json.dumps(rls_rules, sort_keys=True).encode("utf-8")
+        ).hexdigest()
+        return f"guest-{digest}"
+    return str(get_user_id())
+
+
 def apply_rls(
     database: Database,
     catalog: str | None,
@@ -204,6 +227,12 @@ def collect_rls_predicates_for_sql(
             }
         )
     except Exception:
-        # If we can't parse the SQL, return empty list
-        # This ensures RLS application failure doesn't break caching
-        return []
+        # If we can't parse the SQL, we can't tell which (if any) RLS
+        # predicates would apply, so we can't contribute a meaningful cache
+        # key component. Returning an empty list here would make every
+        # user's failure collapse onto the same (missing) contribution,
+        # which is unsafe when different users have different RLS scopes on
+        # the underlying tables. Fall back to a per-user marker instead, so
+        # the cache key still varies by user even though we don't know the
+        # actual predicates.
+        return [f"rls-predicate-parse-failed-for-user-{_get_cache_identity()}"]