fix(trino): catch reserved-word function calls and fix stale sqlglot version note

`_extract_function_calls` only matched TokenType.VAR immediately before `(`,
missing reserved words with their own token type (`current_user`,
`localtime`) that are still callable with parens, letting them slip past
DISALLOWED_SQL_FUNCTIONS if called inside a UDF body. Match on identifier-
shaped token text instead. Also corrects the `_parse` docstring, which still
claimed sqlglot 30.8.0 while requirements/base.txt already pins 30.16.0
(diffed the two, the copy is still faithful).

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

superset/common/query_object.py

@@ -17,6 +17,7 @@
 # pylint: disable=invalid-name
 from __future__ import annotations
 
+import inspect
 import logging
 from datetime import datetime
 from pprint import pformat
@@ -205,8 +206,86 @@ class QueryObject:  # pylint: disable=too-many-instance-attributes
     def _set_post_processing(
         self, post_processing: list[dict[str, Any] | None] | None
     ) -> None:
-        post_processing = post_processing or []
-        self.post_processing = [post_proc for post_proc in post_processing if post_proc]
+        self.post_processing = [
+            self._drop_unsupported_options(post_proc)
+            for post_proc in post_processing or []
+            if post_proc
+        ]
+
+    @staticmethod
+    def _drop_unsupported_options(post_proc: dict[str, Any]) -> dict[str, Any]:
+        """
+        Drop options that the post-processing operation no longer accepts.
+
+        A chart's ``query_context`` is written when the chart is saved and is
+        never rewritten afterwards, while Explore rebuilds the query from
+        ``form_data`` at every render. A chart saved by an older version of
+        Superset can therefore reference an option that has since been removed
+        from the operation. ``exec_post_processing`` passes the stored options
+        as keyword arguments, so that option raises a bare ``TypeError`` on
+        every path that replays the stored ``query_context`` -- the chart data
+        endpoint, alerts and reports, thumbnails, CSV export -- while the same
+        chart still renders correctly in Explore.
+
+        Comparing against the signature avoids a hard-coded list of removed
+        option names, which would need extending at each release.
+        """
+        operation = post_proc.get("operation")
+        function = (
+            getattr(pandas_postprocessing, operation, None)
+            if isinstance(operation, str)
+            else None
+        )
+        if function is None:
+            # A missing or unknown operation is left untouched, so that
+            # exec_post_processing reports it as InvalidPostProcessingError.
+            return post_proc
+
+        parameters = inspect.signature(function).parameters
+        if any(
+            parameter.kind is inspect.Parameter.VAR_KEYWORD
+            for parameter in parameters.values()
+        ):
+            return post_proc
+
+        # `exec_post_processing` calls the operation as `operation(df, **options)`,
+        # so an option can only reach a parameter that a caller may fill by
+        # keyword. That excludes the first parameter, which receives the
+        # DataFrame positionally, and any positional-only or `*args` parameter.
+        keyword_parameters = {
+            name
+            for position, (name, parameter) in enumerate(parameters.items())
+            if position > 0
+            and parameter.kind
+            in (
+                inspect.Parameter.POSITIONAL_OR_KEYWORD,
+                inspect.Parameter.KEYWORD_ONLY,
+            )
+        }
+
+        options = post_proc.get("options") or {}
+        unsupported = {key for key in options if key not in keyword_parameters}
+        if not unsupported:
+            return post_proc
+
+        # Logged at info: a chart saved before the option was removed hits this
+        # on every render, so a warning would repeat for as long as the chart
+        # is not resaved, without anything new to report.
+        logger.info(
+            "Dropping unsupported option(s) %s of post-processing operation "
+            "`%s`. The chart's stored query_context predates the current "
+            "signature of that operation.",
+            sorted(unsupported),
+            operation,
+        )
+        return {
+            **post_proc,
+            "options": {
+                key: value
+                for key, value in options.items()
+                if key in keyword_parameters
+            },
+        }
 
     def _init_series_columns(
         self,

superset/sql/dialects/__init__.py

@@ -21,6 +21,7 @@ from .firebolt import Firebolt, FireboltOld
 from .hana import Hana
 from .opensearch import OpenSearch
 from .pinot import Pinot
+from .trino import Trino
 from .vertica import Vertica
 
 __all__ = [
@@ -31,5 +32,6 @@ __all__ = [
     "Hana",
     "OpenSearch",
     "Pinot",
+    "Trino",
     "Vertica",
 ]

superset/sql/dialects/trino.py

@@ -0,0 +1,469 @@
+# 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 __future__ import annotations
+
+import typing as t
+
+from sqlglot import exp
+from sqlglot.dialects.trino import Trino as SqlglotTrino
+from sqlglot.tokens import Token, TokenType
+
+# Keywords that open a block terminated by ``END`` in Trino SQL routines
+# (https://trino.io/docs/current/udf/sql.html). ``CASE`` is included because
+# both the ``CASE`` statement and the ``CASE`` expression are terminated by
+# ``END``, so counting them keeps the depth balanced either way.
+BLOCK_OPENERS: set[str] = {"BEGIN", "CASE", "IF", "LOOP", "REPEAT", "WHILE"}
+
+# Keywords that are also scalar functions in Trino (e.g. ``IF(a, b, c)`` and
+# ``REPEAT('a', 3)``). When immediately followed by ``(`` they are function
+# calls, not block openers, unless the token stream shows otherwise (see
+# ``_is_paren_condition_block``).
+AMBIGUOUS_OPENERS: set[str] = {"IF", "REPEAT"}
+
+BODY_KEYWORDS: tuple[str, str] = ("RETURN", "BEGIN")
+
+# ``BEGIN``, ``CASE``, and ``END`` are reserved words in sqlglot's Trino
+# tokenizer, so they always carry one of these dedicated token types when
+# used as keywords, and a different one (``STRING``/``IDENTIFIER``) when
+# used as a string literal or quoted identifier, e.g. the string ``'END'``
+# or the quoted identifier ``"end"``. ``IF``, ``LOOP``, ``REPEAT``, and
+# ``WHILE`` are not reserved, so the tokenizer emits ``VAR`` for them both
+# when they're used as a keyword and when they're an unquoted identifier;
+# requiring ``VAR`` still rules out string literals and quoted identifiers,
+# which is the ambiguity ``_is_keyword_token`` guards against.
+_RESERVED_BLOCK_TOKEN_TYPES: dict[str, TokenType] = {
+    "BEGIN": TokenType.BEGIN,
+    "CASE": TokenType.CASE,
+    "END": TokenType.END,
+}
+
+# Token text that can immediately precede a new routine statement inside a
+# ``BEGIN ... END`` body: the start of the body itself, a statement
+# separator, a branch/loop keyword that introduces a nested statement list,
+# or ``:`` following a statement label (e.g. ``top: WHILE ... END WHILE``).
+# Used by ``_is_routine_keyword`` to tell a non-reserved block-opening
+# keyword (``IF``, ``LOOP``, ``REPEAT``, ``WHILE``) apart from an unquoted
+# routine parameter or column reference spelled the same way, since Trino
+# does not reserve these words and its tokenizer emits ``VAR`` for both.
+_STATEMENT_START_PREV_TEXTS: frozenset[str] = frozenset(
+    {"BEGIN", ";", "THEN", "ELSE", "DO", "LOOP", "REPEAT", ":"}
+)
+
+
+def _is_keyword_token(token: Token, text: str) -> bool:
+    """
+    Determine whether ``token`` (whose upper-cased text is ``text``) is an
+    actual occurrence of a routine keyword, as opposed to a string literal
+    or quoted identifier that happens to spell the same word.
+    """
+    if (expected := _RESERVED_BLOCK_TOKEN_TYPES.get(text)) is not None:
+        return token.token_type == expected
+    return token.token_type == TokenType.VAR
+
+
+def _is_routine_keyword(token: Token, text: str, prev_text: str) -> bool:
+    """
+    Determine whether ``token`` is an actual occurrence of a routine block
+    keyword, as opposed to a string literal or quoted identifier that
+    happens to spell the same word (see ``_is_keyword_token``), or, for the
+    non-reserved keywords (``IF``, ``LOOP``, ``REPEAT``, ``WHILE``), an
+    unquoted parameter or column reference spelled the same way, e.g. a UDF
+    parameter named ``loop`` in ``RETURN loop``. A block-opening keyword only
+    ever appears where a new statement can start, so ``prev_text`` (the
+    upper-cased text of the immediately preceding token) is checked against
+    ``_STATEMENT_START_PREV_TEXTS`` for these ambiguous, non-reserved words.
+    """
+    if not _is_keyword_token(token, text):
+        return False
+    if text in _RESERVED_BLOCK_TOKEN_TYPES:
+        return True
+    return prev_text in _STATEMENT_START_PREV_TEXTS
+
+
+def _is_paren_condition_block(tokens: t.Sequence[Token], paren_index: int) -> bool:
+    """
+    Determine whether the parenthesized group starting at ``tokens[paren_index]``
+    (an ``L_PAREN``) is a procedural block condition, e.g. ``IF (a > b) THEN``,
+    as opposed to a scalar function call argument list, e.g. ``IF(a, b, c)``.
+
+    Only ``IF`` has this ambiguity: a parenthesized condition is followed by
+    ``THEN``, while a scalar function call's closing paren never is.
+    """
+    depth = 0
+    for i in range(paren_index, len(tokens)):
+        token_type = tokens[i].token_type
+        if token_type == TokenType.L_PAREN:
+            depth += 1
+        elif token_type == TokenType.R_PAREN:
+            depth -= 1
+            if depth == 0:
+                next_token = tokens[i + 1] if i + 1 < len(tokens) else None
+                return (
+                    next_token is not None and next_token.token_type == TokenType.THEN
+                )
+    return False
+
+
+def _extract_function_calls(tokens: t.Sequence[Token]) -> list[exp.Anonymous]:
+    """
+    Scan the raw tokens of an inline UDF specification for scalar function
+    calls, e.g. ``regexp_replace(...)`` in ``RETURN regexp_replace(...)``, so
+    that ``SQLScript.check_functions_present`` still sees them even though
+    the UDF body itself is kept as opaque, verbatim text.
+
+    A call is any word-like token immediately followed by ``(``. Most scalar
+    functions tokenize as plain ``VAR`` (Trino's tokenizer does not
+    distinguish an unquoted identifier from an unreserved keyword), but a few
+    (e.g. ``current_user``, ``localtime``) are reserved words with their own
+    dedicated ``TokenType`` and would otherwise slip past a ``VAR``-only
+    check while still being callable with parentheses, so the token text
+    itself (rather than its type) decides whether it looks like a call head.
+    This can also match a routine/parameter type name (e.g. ``varchar(10)``),
+    a keyword used with parenthesized syntax (e.g. ``CAST(...)``, ``IN
+    (...)``), or the UDF's own name at its declaration site; those false
+    positives are harmless here, since this list is only used to check for
+    the presence of specific denylisted function names, not to validate the
+    call itself.
+    """
+    return [
+        exp.Anonymous(this=tokens[i - 1].text)
+        for i in range(1, len(tokens))
+        if tokens[i].token_type == TokenType.L_PAREN
+        and tokens[i - 1].text.isidentifier()
+    ]
+
+
+class InlineUDF(exp.CTE):
+    """
+    An inline SQL user-defined function declared in a ``WITH`` clause.
+
+    Trino supports declaring UDFs inline as part of a query::
+
+        WITH FUNCTION meaning_of_life()
+          RETURNS tinyint
+          BEGIN
+            DECLARE a tinyint DEFAULT CAST(6 AS tinyint);
+            DECLARE b tinyint DEFAULT CAST(7 AS tinyint);
+            RETURN a * b;
+          END
+        SELECT meaning_of_life()
+
+    The function definition is stored verbatim as an opaque string (wrapped
+    in an ``exp.Var`` so that AST traversal helpers see an expression), since
+    sqlglot has no representation for SQL routine bodies. Trino does not
+    allow queries inside SQL UDF bodies, so no table references are hidden
+    by the opaque representation. Scalar function calls, however, would be
+    hidden from ``SQLScript.check_functions_present`` (used to enforce
+    ``DISALLOWED_SQL_FUNCTIONS``) since it walks the AST for ``exp.Func``
+    nodes, so those are additionally extracted into ``expressions`` as
+    ``exp.Anonymous`` nodes; they play no part in regenerating the SQL.
+
+    This subclasses ``exp.CTE`` because ``sqlglot.parser.Parser._parse_with``
+    only collects ``exp.CTE`` instances into the ``WITH`` clause.
+    """
+
+    arg_types = {"this": True, "expressions": False}
+
+
+class Trino(SqlglotTrino):
+    """
+    Custom Trino dialect with support for inline SQL UDFs.
+
+    sqlglot cannot parse Trino SQL routine syntax; see
+    https://github.com/tobymao/sqlglot/issues/5178. There are two separate
+    problems:
+
+    1. The parser splits statements on every semicolon, including the ones
+       inside a ``BEGIN ... END`` routine body.
+    2. The ``FUNCTION`` specification in a ``WITH`` clause is not valid CTE
+       syntax.
+
+    This dialect keeps routine bodies intact when splitting statements, and
+    parses inline function specifications into opaque `InlineUDF` nodes that
+    regenerate verbatim.
+
+    Note that sqlglot's ``Dialect`` metaclass registers subclasses by class
+    name, so once this module is imported this class also replaces the
+    built-in dialect for string-based lookups (``dialect="trino"``). This is
+    intentional, and consistent with how other Superset dialects (e.g.
+    ``Dremio``) shadow their sqlglot counterparts: the extensions are purely
+    additive, only activating on syntax that fails to parse upstream.
+    """
+
+    class Parser(SqlglotTrino.Parser):
+        @staticmethod
+        def _block_depth_delta(
+            tokens: list[Token],
+            index: int,
+            prev_text: str,
+        ) -> int:
+            """
+            Compute the block nesting change contributed by the routine token
+            at ``tokens[index]``.
+            """
+            token = tokens[index]
+            text = token.text.upper()
+            if text in BLOCK_OPENERS:
+                if not _is_routine_keyword(token, text, prev_text):
+                    return 0  # literal, identifier, or parameter reference
+                if prev_text == "END":
+                    return 0  # block terminator, e.g. `END IF`, `END CASE`
+                next_token = tokens[index + 1] if index + 1 < len(tokens) else None
+                if (
+                    text in AMBIGUOUS_OPENERS
+                    and next_token
+                    and next_token.token_type == TokenType.L_PAREN
+                ):
+                    if text == "IF" and _is_paren_condition_block(tokens, index + 1):
+                        return 1  # procedural `IF (...) THEN`, not a call
+                    return 0  # scalar function call, e.g. `IF(a, b, c)`
+                return 1
+            if text == "END" and _is_routine_keyword(token, text, prev_text):
+                return -1
+            return 0
+
+        @staticmethod
+        def _starts_routine(
+            heads: list[TokenType],
+            next_token_type: TokenType | None,
+            paren_depth: int,
+        ) -> bool:
+            """
+            Determine whether a ``FUNCTION`` token at the end of ``heads``
+            (excluded from the list) begins a new routine specification:
+            ``CREATE FUNCTION``, ``CREATE OR REPLACE FUNCTION``, or an entry
+            in a ``WITH`` list, either right after ``WITH`` itself or after a
+            top-level comma separating it from a preceding CTE, e.g.
+            ``WITH cte AS (...), FUNCTION f() ...``.
+
+            In the ``WITH`` case, ``FUNCTION`` may also just be an ordinary
+            CTE named "function", e.g. ``WITH function AS (...) SELECT ...``.
+            ``next_token_type`` (the token immediately after ``FUNCTION``) is
+            checked the same way ``_parse_cte`` disambiguates the two: a CTE
+            named "function" is followed by ``AS``, ``(``, or a comma (for a
+            column alias list), while a routine specification is followed by
+            the function name.
+            """
+            if heads[:1] == [TokenType.CREATE]:
+                return heads in (
+                    [TokenType.CREATE],
+                    [TokenType.CREATE, TokenType.OR, TokenType.REPLACE],
+                )
+            if heads[:1] == [TokenType.WITH]:
+                return (
+                    paren_depth == 0
+                    and heads[-1] in (TokenType.WITH, TokenType.COMMA)
+                    and next_token_type
+                    not in (TokenType.ALIAS, TokenType.L_PAREN, TokenType.COMMA)
+                )
+            return False
+
+        def _parse(
+            self,
+            parse_method: t.Callable[..., exp.Expression | None],
+            raw_tokens: list[Token],
+            sql: str | None = None,
+        ) -> list[exp.Expression | None]:
+            """
+            Split tokens into statements, keeping routine bodies intact.
+
+            This is a copy of ``sqlglot.parser.Parser._parse`` (verified to
+            match through sqlglot 30.16.0, the version pinned in
+            ``requirements/base.txt`` as of this writing) with one change:
+            when a statement starts with ``WITH FUNCTION``, ``CREATE
+            FUNCTION``, or ``CREATE OR REPLACE FUNCTION``, semicolons inside
+            ``BEGIN ... END`` blocks do not split the statement. Because this
+            is a hand-maintained copy rather than an extension through a
+            public hook, it will silently drift if sqlglot's own ``_parse``
+            changes on a future upgrade; re-diff this method against the new
+            version whenever ``sqlglot`` is bumped in ``requirements/base.txt``.
+            """
+            self.reset()
+            self.sql = sql or ""
+
+            total = len(raw_tokens)
+            chunks: list[list[Token]] = [[]]
+            routine_mode: bool = False
+            depth: int = 0
+            paren_depth: int = 0
+            prev_text: str = ""
+
+            for i, token in enumerate(raw_tokens):
+                if token.token_type == TokenType.SEMICOLON and depth <= 0:
+                    if token.comments:
+                        chunks.append([token])
+                    if i < total - 1:
+                        chunks.append([])
+                    routine_mode = False
+                    depth = 0
+                    paren_depth = 0
+                    prev_text = ""
+                    continue
+
+                chunk = chunks[-1]
+                chunk.append(token)
+
+                if token.token_type == TokenType.FUNCTION and not routine_mode:
+                    heads = [tok.token_type for tok in chunk[:-1]]
+                    next_token = raw_tokens[i + 1] if i + 1 < total else None
+                    routine_mode = self._starts_routine(
+                        heads,
+                        next_token.token_type if next_token else None,
+                        paren_depth,
+                    )
+                elif routine_mode:
+                    depth += self._block_depth_delta(raw_tokens, i, prev_text)
+
+                if token.token_type == TokenType.L_PAREN:
+                    paren_depth += 1
+                elif token.token_type == TokenType.R_PAREN:
+                    paren_depth -= 1
+
+                prev_text = token.text.upper()
+
+            self._chunks = chunks
+            return self._parse_batch_statements(
+                parse_method=parse_method,
+                sep_first_statement=False,
+            )
+
+        def _parse_cte(self) -> exp.CTE | None:
+            """
+            Parse a single entry in a ``WITH`` clause.
+
+            An entry starting with the ``FUNCTION`` keyword followed by an
+            identifier is an inline UDF specification; anything else
+            (including a CTE named "function") is handled by sqlglot.
+            """
+            if (
+                self._curr
+                and self._curr.token_type == TokenType.FUNCTION
+                and self._next
+                and self._next.token_type
+                not in (TokenType.ALIAS, TokenType.L_PAREN, TokenType.COMMA)
+            ):
+                return self._parse_inline_udf()
+
+            return super()._parse_cte()
+
+        def _parse_inline_udf(self) -> InlineUDF:
+            """
+            Consume an inline UDF specification and return it verbatim.
+
+            The specification is ``FUNCTION name(params) RETURNS type`` plus
+            optional routine characteristics, followed by a body that is
+            either ``RETURN expression`` or a ``BEGIN ... END`` block.
+            """
+            start = self._curr
+            start_index = self._index
+            self._advance()
+
+            # scan for the start of the function body, skipping over the
+            # signature, return type, and routine characteristics. The
+            # ``_is_keyword_token`` check rules out a routine characteristic
+            # whose string value happens to spell a body keyword, e.g.
+            # ``COMMENT 'RETURN'`` or ``COMMENT 'BEGIN'``.
+            paren_depth: int = 0
+            body: str | None = None
+            while self._curr:
+                token_type = self._curr.token_type
+                text = self._curr.text.upper()
+                if token_type == TokenType.L_PAREN:
+                    paren_depth += 1
+                elif token_type == TokenType.R_PAREN:
+                    paren_depth -= 1
+                elif (
+                    paren_depth == 0
+                    and text in BODY_KEYWORDS
+                    and _is_keyword_token(self._curr, text)
+                ):
+                    body = text
+                    break
+                self._advance()
+
+            if body is None:
+                self.raise_error(
+                    "Expected RETURN or BEGIN in inline function specification"
+                )
+
+            if body == "RETURN":
+                self._advance()
+                if not self._parse_expression():
+                    self.raise_error("Expected expression after RETURN")
+            else:
+                self._consume_block()
+
+            raw = self.sql[start.start : self._prev.end + 1]
+            calls = _extract_function_calls(self._tokens[start_index : self._index])
+            return self.expression(
+                InlineUDF(this=exp.Var(this=raw), expressions=calls), token=start
+            )
+
+        def _consume_block(self) -> None:
+            """
+            Consume a ``BEGIN ... END`` block, tracking nested blocks.
+            """
+            depth: int = 0
+            prev_text: str = ""
+            while self._curr:
+                token = self._curr
+                text = token.text.upper()
+                if text in BLOCK_OPENERS and _is_routine_keyword(
+                    token, text, prev_text
+                ):
+                    is_scalar_call = (
+                        text in AMBIGUOUS_OPENERS
+                        and self._next
+                        and self._next.token_type == TokenType.L_PAREN
+                        and not (
+                            text == "IF"
+                            and _is_paren_condition_block(self._tokens, self._index + 1)
+                        )
+                    )
+                    if is_scalar_call:
+                        pass  # scalar function call, e.g. `IF(a, b, c)`
+                    else:
+                        depth += 1
+                    prev_text = text
+                    self._advance()
+                elif text == "END" and _is_routine_keyword(token, text, prev_text):
+                    depth -= 1
+                    prev_text = text
+                    self._advance()
+                    if (
+                        depth > 0
+                        and self._curr
+                        and self._curr.text.upper() in BLOCK_OPENERS
+                        and _is_keyword_token(self._curr, self._curr.text.upper())
+                    ):
+                        # block terminator, e.g. `END IF`, `END CASE`
+                        prev_text = self._curr.text.upper()
+                        self._advance()
+                    if depth == 0:
+                        return
+                else:
+                    prev_text = text
+                    self._advance()
+
+            self.raise_error("Unbalanced BEGIN/END in inline function specification")
+
+    class Generator(SqlglotTrino.Generator):
+        TRANSFORMS = {
+            **SqlglotTrino.Generator.TRANSFORMS,
+            InlineUDF: lambda self, e: e.this.name,
+        }

superset/sql/parse.py

@@ -55,6 +55,7 @@ from superset.sql.dialects import (
     Hana,
     OpenSearch,
     Pinot,
+    Trino,
     Vertica,
 )
 
@@ -161,7 +162,7 @@ SQLGLOT_DIALECTS = {
     "superset": Dialects.SQLITE,
     # "taosws": ???
     "teradatasql": Dialects.TERADATA,
-    "trino": Dialects.TRINO,
+    "trino": Trino,
     "vertica": Vertica,
     # "ydb" is a plugin dialect (ydb-sqlglot-plugin) auto-discovered via entry_points,
     # hence a string name rather than a class reference like the built-in dialects.

superset/utils/pandas_postprocessing/utils.py

@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 from collections.abc import Sequence
-from functools import partial
+from functools import partial, wraps
 from typing import Any, Callable
 
 import numpy as np
@@ -122,6 +122,10 @@ def scalar_to_sequence(val: Any) -> Sequence[str]:
 
 def validate_column_args(*argnames: str) -> Callable[..., Any]:
     def wrapper(func: Callable[..., Any]) -> Callable[..., Any]:
+        # `wraps` keeps `func` reachable through `__wrapped__`, so that
+        # `inspect.signature` reports the parameters of the decorated operation
+        # rather than the `(df, **options)` of this wrapper.
+        @wraps(func)
         def wrapped(df: DataFrame, **options: Any) -> Any:
             if _is_multi_index_on_columns(df):
                 # MultiIndex column validate first level

tests/unit_tests/pandas_postprocessing/test_utils.py

@@ -14,7 +14,13 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from superset.utils.pandas_postprocessing import escape_separator, unescape_separator
+import inspect
+
+from superset.utils.pandas_postprocessing import (
+    escape_separator,
+    pivot,
+    unescape_separator,
+)
 
 
 def test_escape_separator():
@@ -28,3 +34,19 @@ def test_escape_separator():
     escape_string = escape_separator("hello,world")
     assert escape_string == r"hello\,world"
     assert unescape_separator(escape_string) == "hello,world"
+
+
+def test_validate_column_args_preserves_signature():
+    """
+    The decorator must not hide the signature of the operation it wraps.
+
+    `inspect.signature` follows `__wrapped__`, which `functools.wraps` sets.
+    Without it every decorated operation reports `(df, **options)`, and code
+    that inspects the signature -- see `QueryObject._drop_unsupported_options`
+    -- cannot tell a supported option from an unsupported one.
+    """
+    parameters = inspect.signature(pivot).parameters
+
+    assert pivot.__name__ == "pivot"
+    assert "options" not in parameters
+    assert {"index", "aggregates", "columns"} <= set(parameters)

tests/unit_tests/queries/query_object_test.py

@@ -22,6 +22,7 @@ from superset.common.query_object import QueryObject
 from superset.connectors.sqla.models import SqlaTable
 from superset.models.core import Database
 from superset.superset_typing import Metric
+from superset.utils import pandas_postprocessing
 from superset.utils.core import override_user
 
 
@@ -438,3 +439,143 @@ def test_cache_key_cache_impersonation_on_with_different_user_and_db_impersonati
         ],
         any_order=True,
     )
+
+
+def test_post_processing_drops_unsupported_options():
+    """
+    An option that the operation no longer accepts is dropped, not passed on.
+
+    A chart saved by an older version of Superset stores `flatten_columns` in
+    the options of its `pivot` operation. `pivot` lost that parameter when
+    flattening became its own operation, so replaying the stored query_context
+    raised `TypeError: pivot() got an unexpected keyword argument
+    'flatten_columns'`.
+    """
+    query_object = QueryObject(
+        row_limit=1,
+        post_processing=[
+            {
+                "operation": "pivot",
+                "options": {
+                    "index": ["__timestamp"],
+                    "columns": ["genre"],
+                    "aggregates": {"count": {"operator": "mean"}},
+                    "drop_missing_columns": False,
+                    "flatten_columns": True,
+                    "reset_index": True,
+                },
+            }
+        ],
+    )
+
+    options = query_object.post_processing[0]["options"]
+    assert "flatten_columns" not in options
+    assert "reset_index" not in options
+    assert options["drop_missing_columns"] is False
+    assert options["index"] == ["__timestamp"]
+
+
+def test_post_processing_keeps_supported_options():
+    """Options the operation accepts are left alone."""
+    post_processing = [
+        {
+            "operation": "pivot",
+            "options": {"index": ["__timestamp"], "aggregates": {}},
+        }
+    ]
+    query_object = QueryObject(row_limit=1, post_processing=post_processing)
+
+    assert query_object.post_processing == post_processing
+
+
+def test_post_processing_keeps_unknown_operation():
+    """
+    An unknown operation is kept, so that `exec_post_processing` can report it
+    as an `InvalidPostProcessingError` rather than being silently dropped here.
+    """
+    query_object = QueryObject(
+        row_limit=1,
+        post_processing=[{"operation": "does_not_exist", "options": {"a": 1}}, None],
+    )
+
+    assert query_object.post_processing == [
+        {"operation": "does_not_exist", "options": {"a": 1}}
+    ]
+
+
+def test_post_processing_drops_the_dataframe_parameter():
+    """
+    The DataFrame parameter is not an option.
+
+    `exec_post_processing` calls `operation(df, **options)`, so an option named
+    after the first parameter would raise `TypeError: pivot() got multiple
+    values for argument 'df'`.
+    """
+    query_object = QueryObject(
+        row_limit=1,
+        post_processing=[
+            {
+                "operation": "pivot",
+                "options": {"df": "malformed", "index": ["a"], "aggregates": {}},
+            }
+        ],
+    )
+
+    options = query_object.post_processing[0]["options"]
+    assert "df" not in options
+    assert options["index"] == ["a"]
+
+
+def test_post_processing_keeps_options_of_a_variadic_operation():
+    """An operation that accepts `**kwargs` accepts every option."""
+
+    def variadic(df, **kwargs):
+        return df
+
+    post_processing = [{"operation": "variadic", "options": {"anything": 1}}]
+    with patch.object(pandas_postprocessing, "variadic", variadic, create=True):
+        query_object = QueryObject(row_limit=1, post_processing=post_processing)
+
+    assert query_object.post_processing == post_processing
+
+
+def test_post_processing_drops_a_variadic_positional_option():
+    """
+    A `*args` parameter cannot be filled by a keyword argument.
+
+    `exec_post_processing` calls the operation as `operation(df, **options)`,
+    so an option named after a `*args` parameter would raise `TypeError:
+    variadic_positional() got an unexpected keyword argument 'args'` even
+    though the name appears in the signature.
+    """
+
+    def variadic_positional(df, *args, index=None):  # pylint: disable=unused-argument
+        return df
+
+    with patch.object(
+        pandas_postprocessing, "variadic_positional", variadic_positional, create=True
+    ):
+        query_object = QueryObject(
+            row_limit=1,
+            post_processing=[
+                {
+                    "operation": "variadic_positional",
+                    "options": {"args": [1], "index": ["a"]},
+                }
+            ],
+        )
+
+    options = query_object.post_processing[0]["options"]
+    assert "args" not in options
+    assert options["index"] == ["a"]
+
+
+def test_post_processing_keeps_an_entry_without_an_operation():
+    """
+    An entry that names no operation is kept, so that `exec_post_processing`
+    reports it as an `InvalidPostProcessingError`.
+    """
+    post_processing = [{"options": {"a": 1}}]
+    query_object = QueryObject(row_limit=1, post_processing=post_processing)
+
+    assert query_object.post_processing == post_processing

tests/unit_tests/sql/dialects/trino_tests.py

@@ -0,0 +1,520 @@
+# 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 pytest
+import sqlglot
+
+from superset.exceptions import SupersetParseError
+from superset.sql.dialects.trino import InlineUDF, Trino
+from superset.sql.parse import SQLScript, SQLStatement, Table
+
+# example from https://trino.io/docs/current/udf/sql/begin.html, reported in
+# https://github.com/apache/superset/issues/26162
+INLINE_UDF = """
+WITH FUNCTION meaning_of_life()
+  RETURNS tinyint
+  BEGIN
+    DECLARE a tinyint DEFAULT CAST(6 AS tinyint);
+    DECLARE b tinyint DEFAULT CAST(7 AS tinyint);
+    RETURN a * b;
+  END
+SELECT meaning_of_life()
+""".strip()
+
+
+def test_inline_udf_is_single_statement() -> None:
+    """
+    Semicolons inside the routine body must not split the statement.
+    """
+    statements = sqlglot.parse(INLINE_UDF, dialect=Trino)
+    assert len(statements) == 1
+    assert len(list(statements[0].find_all(InlineUDF))) == 1
+
+
+def test_inline_udf_generates_verbatim() -> None:
+    """
+    The function specification should be preserved verbatim, and the
+    generated SQL should be parseable again.
+    """
+    statement = sqlglot.parse_one(INLINE_UDF, dialect=Trino)
+    generated = statement.sql(dialect=Trino)
+    assert (
+        """
+WITH FUNCTION meaning_of_life()
+  RETURNS tinyint
+  BEGIN
+    DECLARE a tinyint DEFAULT CAST(6 AS tinyint);
+    DECLARE b tinyint DEFAULT CAST(7 AS tinyint);
+    RETURN a * b;
+  END
+""".strip()
+        in generated
+    )
+    assert sqlglot.parse_one(generated, dialect=Trino)
+
+
+def test_inline_udf_return_form() -> None:
+    """
+    Test functions whose body is a single ``RETURN`` expression, including
+    multiple comma-separated functions in one ``WITH`` clause.
+    """
+    sql = """
+WITH
+  FUNCTION hello(name varchar)
+    RETURNS varchar
+    RETURN format('Hello %s!', name),
+  FUNCTION bye()
+    RETURNS varchar
+    RETURN 'Bye!'
+SELECT hello('Finn') || ' and ' || bye()
+    """.strip()
+    statement = sqlglot.parse_one(sql, dialect=Trino)
+    assert len(list(statement.find_all(InlineUDF))) == 2
+
+    generated = statement.sql(dialect=Trino)
+    assert "RETURN format('Hello %s!', name)" in generated
+    assert "RETURN 'Bye!'" in generated
+
+
+@pytest.mark.parametrize(
+    "sql",
+    [
+        """
+WITH FUNCTION classify(a bigint)
+  RETURNS varchar
+  BEGIN
+    CASE a
+      WHEN 0 THEN RETURN 'zero';
+      WHEN 1 THEN RETURN 'one';
+      ELSE RETURN 'more than one or negative';
+    END CASE;
+    RETURN NULL;
+  END
+SELECT classify(x) FROM some_table
+        """,
+        """
+WITH FUNCTION classify(a bigint)
+  RETURNS varchar
+  BEGIN
+    IF a > 100 THEN
+      RETURN 'big';
+    ELSEIF a > 0 THEN
+      RETURN 'small';
+    END IF;
+    RETURN 'negative';
+  END
+SELECT classify(x) FROM some_table
+        """,
+        """
+WITH FUNCTION classify(a bigint)
+  RETURNS varchar
+  BEGIN
+    WHILE a < 100 DO
+      SET a = a + 1;
+    END WHILE;
+    RETURN IF(a = 100, 'hundred', 'other');
+  END
+SELECT classify(x) FROM some_table
+        """,
+        """
+WITH FUNCTION classify(a bigint)
+  RETURNS varchar
+  BEGIN
+    IF (a > 100) THEN
+      RETURN 'big';
+    ELSEIF a > 0 THEN
+      RETURN 'small';
+    END IF;
+    RETURN 'negative';
+  END
+SELECT classify(x) FROM some_table
+        """,
+    ],
+)
+def test_inline_udf_nested_blocks(sql: str) -> None:
+    """
+    Test nested blocks: ``CASE ... END CASE``, ``IF ... END IF``,
+    ``WHILE ... END WHILE``, scalar ``IF()`` function calls, and a
+    parenthesized ``IF (...) THEN`` condition.
+    """
+    statements = sqlglot.parse(sql.strip(), dialect=Trino)
+    assert len(statements) == 1
+
+
+def test_cte_named_function_still_works() -> None:
+    """
+    A CTE named "function" must still be parsed as a regular CTE.
+    """
+    sql = "WITH function AS (SELECT 1 AS x) SELECT x FROM function"
+    statement = sqlglot.parse_one(sql, dialect=Trino)
+    assert not list(statement.find_all(InlineUDF))
+    assert statement.sql(dialect=Trino) == sql
+
+
+def test_inline_udf_after_regular_cte() -> None:
+    """
+    An inline UDF following a regular CTE in the same ``WITH`` clause must
+    still have its body's semicolons kept intact.
+    """
+    sql = """
+WITH cte AS (SELECT 1),
+FUNCTION meaning_of_life()
+  RETURNS tinyint
+  BEGIN
+    DECLARE a tinyint DEFAULT CAST(6 AS tinyint);
+    DECLARE b tinyint DEFAULT CAST(7 AS tinyint);
+    RETURN a * b;
+  END
+SELECT meaning_of_life()
+""".strip()
+    statements = sqlglot.parse(sql, dialect=Trino)
+    assert len(statements) == 1
+    assert len(list(statements[0].find_all(InlineUDF))) == 1
+
+
+def test_unbalanced_body_raises() -> None:
+    """
+    An unterminated routine body should raise a parse error.
+    """
+    sql = "WITH FUNCTION f() RETURNS int BEGIN RETURN 1; SELECT f()"
+    with pytest.raises(sqlglot.errors.ParseError):
+        sqlglot.parse(sql, dialect=Trino)
+
+
+def test_missing_body_raises() -> None:
+    """
+    A function specification without a body should raise a parse error.
+    """
+    sql = "WITH FUNCTION f() RETURNS int SELECT f()"
+    with pytest.raises(sqlglot.errors.ParseError):
+        sqlglot.parse(sql, dialect=Trino)
+
+
+def test_missing_return_expression_raises() -> None:
+    """
+    A ``RETURN`` body without a following expression should raise a parse
+    error.
+    """
+    sql = "WITH FUNCTION f() RETURNS int RETURN"
+    with pytest.raises(sqlglot.errors.ParseError):
+        sqlglot.parse(sql, dialect=Trino)
+
+
+def test_semicolon_with_trailing_comment() -> None:
+    """
+    A statement-separating semicolon with a comment attached to it (no
+    whitespace in between) should still split statements correctly.
+    """
+    sql = "SELECT 1;-- trailing\nSELECT 2"
+    statements = sqlglot.parse(sql, dialect=Trino)
+    assert len(statements) == 3  # SELECT 1, the comment-bearing `;`, SELECT 2
+
+
+def test_trailing_semicolon_with_no_following_statement() -> None:
+    """
+    A single statement terminated by a semicolon with nothing after it
+    should parse as one statement.
+    """
+    statements = sqlglot.parse("SELECT 1;", dialect=Trino)
+    assert len(statements) == 1
+
+
+def test_sqlscript_inline_udf() -> None:
+    """
+    Integration with the Superset parsing API (reproduces #26162).
+    """
+    script = SQLScript(INLINE_UDF, "trino")
+    assert len(script.statements) == 1
+    assert not script.has_mutation()
+
+    statement = script.statements[0]
+    assert statement.is_select()
+    assert statement.format() == statement.format()  # deterministic
+
+
+def test_sqlscript_inline_udf_multiple_statements() -> None:
+    """
+    Statements after the UDF query should still be split correctly.
+    """
+    script = SQLScript(f"{INLINE_UDF};\nSELECT 42", "trino")
+    assert len(script.statements) == 2
+
+
+def test_sqlstatement_extract_tables() -> None:
+    """
+    Tables referenced by the main query should still be extracted.
+    """
+    sql = """
+WITH FUNCTION doubleup(x integer)
+  RETURNS integer
+  BEGIN
+    RETURN x * 2;
+  END
+SELECT doubleup(some_column) FROM some_table
+    """.strip()
+    statement = SQLStatement(sql, "trino")
+    assert statement.tables == {Table("some_table")}
+
+
+def test_sqlstatement_regular_queries_unaffected() -> None:
+    """
+    Regular Trino queries should parse exactly as before.
+    """
+    script = SQLScript(
+        "WITH t AS (SELECT 1 AS x) SELECT * FROM t; SELECT 2",
+        "trino",
+    )
+    assert len(script.statements) == 2
+    assert script.statements[0].tables == set()
+
+    with pytest.raises(SupersetParseError):
+        SQLStatement("SELECT * FROM", "trino")
+
+
+def test_inline_udf_nested_parens_in_condition() -> None:
+    """
+    A parenthesized ``IF`` condition containing its own nested parens must
+    still be recognized as a block opener, not a scalar function call.
+    """
+    sql = """
+WITH FUNCTION classify(a bigint, b bigint)
+  RETURNS varchar
+  BEGIN
+    IF ((a > 100) AND (b > 100)) THEN
+      RETURN 'big';
+    END IF;
+    RETURN 'small';
+  END
+SELECT classify(x, y) FROM some_table
+    """.strip()
+    statements = sqlglot.parse(sql, dialect=Trino)
+    assert len(statements) == 1
+    assert len(list(statements[0].find_all(InlineUDF))) == 1
+
+
+def test_scalar_function_named_function() -> None:
+    """
+    A regular scalar function call literally named ``function`` (outside a
+    ``CREATE``/``WITH`` routine specification) must parse normally.
+    """
+    sql = "SELECT function(x) FROM t"
+    statements = sqlglot.parse(sql, dialect=Trino)
+    assert len(statements) == 1
+
+
+def test_unclosed_if_condition_raises() -> None:
+    """
+    An ``IF`` condition with an unbalanced opening paren should fail to
+    parse rather than being silently misread as a block.
+    """
+    sql = (
+        "WITH FUNCTION f() RETURNS int BEGIN "
+        "IF (a > 1 THEN RETURN 1; END IF; RETURN 2; END SELECT 1"
+    )
+    with pytest.raises(sqlglot.errors.ParseError):
+        sqlglot.parse(sql, dialect=Trino)
+
+
+def test_create_function_not_split() -> None:
+    """
+    ``CREATE FUNCTION`` bodies should not be split on semicolons either.
+    """
+    sql = """
+CREATE FUNCTION meaning_of_life()
+  RETURNS tinyint
+  BEGIN
+    DECLARE a tinyint DEFAULT CAST(6 AS tinyint);
+    DECLARE b tinyint DEFAULT CAST(7 AS tinyint);
+    RETURN a * b;
+  END;
+SELECT 42
+    """.strip()
+    statements = sqlglot.parse(sql, dialect=Trino)
+    assert len(statements) == 2
+
+
+def test_create_or_replace_function_not_split() -> None:
+    """
+    ``CREATE OR REPLACE FUNCTION`` bodies should not be split on semicolons
+    either, and the routine is followed by the next statement.
+    """
+    sql = """
+CREATE OR REPLACE FUNCTION meaning_of_life()
+  RETURNS tinyint
+  BEGIN
+    DECLARE a tinyint DEFAULT CAST(6 AS tinyint);
+    DECLARE b tinyint DEFAULT CAST(7 AS tinyint);
+    RETURN a * b;
+  END;
+SELECT 42
+    """.strip()
+    statements = sqlglot.parse(sql, dialect=Trino)
+    assert len(statements) == 2
+
+
+def test_block_keyword_as_parameter_reference_not_counted() -> None:
+    """
+    ``LOOP``, ``REPEAT``, and ``WHILE`` are not reserved words in Trino, so
+    an unquoted routine parameter or column reference spelled the same way
+    (e.g. a parameter named ``loop``) must not be mistaken for a
+    block-opening keyword, which would otherwise leave the block depth
+    unbalanced at ``END``.
+    """
+    sql = """
+WITH FUNCTION echo(loop bigint)
+  RETURNS bigint
+  BEGIN
+    RETURN loop;
+  END
+SELECT echo(x) FROM some_table
+    """.strip()
+    statements = sqlglot.parse(sql, dialect=Trino)
+    assert len(statements) == 1
+    assert len(list(statements[0].find_all(InlineUDF))) == 1
+
+
+def test_body_keyword_in_routine_characteristic_ignored() -> None:
+    """
+    A routine characteristic string value that happens to spell a body
+    keyword (e.g. ``COMMENT 'RETURN'`` or ``COMMENT 'BEGIN'``) must not be
+    mistaken for the actual start of the function body.
+    """
+    sql = """
+WITH FUNCTION f()
+  RETURNS int
+  COMMENT 'RETURN'
+  BEGIN
+    RETURN 1;
+  END
+SELECT f()
+    """.strip()
+    statements = sqlglot.parse(sql, dialect=Trino)
+    assert len(statements) == 1
+    assert len(list(statements[0].find_all(InlineUDF))) == 1
+
+    sql_begin_comment = """
+WITH FUNCTION f()
+  RETURNS int
+  COMMENT 'BEGIN'
+  RETURN 1
+SELECT f()
+    """.strip()
+    statements = sqlglot.parse(sql_begin_comment, dialect=Trino)
+    assert len(statements) == 1
+    assert len(list(statements[0].find_all(InlineUDF))) == 1
+
+
+def test_block_keywords_in_string_literals_and_identifiers_ignored() -> None:
+    """
+    Block keywords (``BEGIN``, ``CASE``, ``END``, ``IF``, ...) that appear as
+    the text of a string literal or a quoted identifier must not be mistaken
+    for actual routine keywords when tracking block depth, since they carry
+    the same text but a different token type.
+    """
+    sql = """
+WITH FUNCTION describe_status(status varchar)
+  RETURNS varchar
+  BEGIN
+    IF status = 'END' THEN
+      RETURN 'terminal';
+    END IF;
+    RETURN "case";
+  END
+SELECT describe_status('END')
+    """.strip()
+    statements = sqlglot.parse(sql, dialect=Trino)
+    assert len(statements) == 1
+    assert len(list(statements[0].find_all(InlineUDF))) == 1
+
+
+def test_cte_named_function_does_not_trigger_routine_mode() -> None:
+    """
+    An ordinary CTE named "function" must not put the parser into routine
+    mode: block keywords used as ordinary identifiers/expressions elsewhere
+    in the script (here, `loop` as a column alias, and the `CASE ... END`
+    expression) must not affect statement splitting, and a later statement
+    must still be split off correctly.
+    """
+    sql = (
+        "WITH function AS (SELECT 1 AS a, 2 AS loop) "
+        "SELECT CASE WHEN a THEN loop ELSE 0 END FROM function; "
+        "SELECT 2"
+    )
+    statements = sqlglot.parse(sql, dialect=Trino)
+    assert len(statements) == 2
+    assert not list(statements[0].find_all(InlineUDF))
+
+
+def test_labeled_loop_block_depth_tracked() -> None:
+    """
+    A labeled loop (``label: WHILE ... END WHILE``, per
+    https://trino.io/docs/current/udf/sql.html) must still be tracked for
+    block depth: the label's trailing ``:`` sits between the loop opener and
+    its preceding statement separator/branch keyword.
+    """
+    sql = """
+WITH FUNCTION count_to(n bigint)
+  RETURNS bigint
+  BEGIN
+    DECLARE r bigint DEFAULT 0;
+    top: WHILE r < n DO
+      SET r = r + 1;
+    END WHILE;
+    RETURN r;
+  END
+SELECT count_to(5)
+    """.strip()
+    statements = sqlglot.parse(sql, dialect=Trino)
+    assert len(statements) == 1
+    assert len(list(statements[0].find_all(InlineUDF))) == 1
+
+
+def test_udf_body_function_calls_visible_to_check_functions_present() -> None:
+    """
+    A scalar function call inside an inline UDF body must still be visible
+    to ``SQLScript.check_functions_present`` (used to enforce
+    ``DISALLOWED_SQL_FUNCTIONS``), even though the body itself is stored as
+    opaque, verbatim text.
+    """
+    sql = """
+WITH FUNCTION mask(x varchar)
+  RETURNS varchar
+  RETURN regexp_replace(x, '.', '*')
+SELECT mask(some_column) FROM some_table
+    """.strip()
+    script = SQLScript(sql, "trino")
+    assert script.statements[0].check_functions_present({"regexp_replace"})
+    assert not script.statements[0].check_functions_present({"not_present"})
+
+
+def test_udf_body_reserved_word_function_call_visible_to_check_functions_present() -> (
+    None
+):
+    """
+    A handful of scalar functions (``current_user``, ``localtime``, etc.) are
+    reserved words with their own dedicated token type rather than the
+    generic ``VAR`` most function names get, so they must still be caught
+    when called with parentheses inside an inline UDF body.
+    """
+    sql = """
+WITH FUNCTION whoami()
+  RETURNS varchar
+  RETURN current_user()
+SELECT whoami()
+    """.strip()
+    script = SQLScript(sql, "trino")
+    assert script.statements[0].check_functions_present({"current_user"})

tests/unit_tests/sql/parse_tests.py

@@ -2013,6 +2013,43 @@ def test_is_mutating_postgres_command_constructs(sql: str, expected: bool) -> No
     assert SQLStatement(sql, "postgresql").is_mutating() == expected
 
 
+@pytest.mark.parametrize(
+    "sql, expected",
+    [
+        # A persistent catalog function has no structured sqlglot grammar and
+        # falls back to an opaque exp.Command("CREATE"), which the generic
+        # exp.Create check does not catch. Without the Trino-specific
+        # exp.Command check, this would slip past a read-only (allow_dml=False)
+        # gate and still create a function on the Trino cluster.
+        (
+            "CREATE FUNCTION meaning_of_life() RETURNS tinyint BEGIN RETURN 42; END",
+            True,
+        ),
+        (
+            "CREATE OR REPLACE FUNCTION meaning_of_life() RETURNS tinyint "
+            "BEGIN RETURN 42; END",
+            True,
+        ),
+        # An inline `WITH FUNCTION` UDF is scoped to the query and does not
+        # persist anything server-side, so it must stay non-mutating.
+        (
+            "WITH FUNCTION meaning_of_life() RETURNS tinyint "
+            "BEGIN RETURN 42; END "
+            "SELECT meaning_of_life()",
+            False,
+        ),
+    ],
+)
+def test_is_mutating_trino_create_function(sql: str, expected: bool) -> None:
+    """
+    Trino `CREATE [OR REPLACE] FUNCTION ... BEGIN ... END` creates a
+    persistent catalog function and must be classified as mutating, even
+    though sqlglot represents it as an opaque `exp.Command` rather than a
+    structured `exp.Create` node.
+    """
+    assert SQLStatement(sql, "trino").is_mutating() == expected
+
+
 @pytest.mark.parametrize(
     "sql, engine, functions, expected",
     [