diff --git a/superset/config.py b/superset/config.py
index 43155366c5ece..2febf07acd254 100644
--- a/superset/config.py
+++ b/superset/config.py
@@ -856,7 +856,7 @@ class CeleryConfig:  # pylint: disable=too-few-public-methods
 # The use case is can be around adding some sort of comment header
 # with information such as the username and worker node information
 #
-#    def SQL_QUERY_MUTATOR(sql, username, security_manager):
+#    def SQL_QUERY_MUTATOR(sql, user_name, security_manager, database):
 #        dttm = datetime.now().isoformat()
 #        return f"-- [SQL LAB] {username} {dttm}\n{sql}"
 SQL_QUERY_MUTATOR = None
diff --git a/superset/sql_lab.py b/superset/sql_lab.py
index 1153a2b7d6be6..9cd3a29bb5770 100644
--- a/superset/sql_lab.py
+++ b/superset/sql_lab.py
@@ -34,8 +34,10 @@
 from superset.dataframe import df_to_records
 from superset.db_engine_specs import BaseEngineSpec
 from superset.extensions import celery_app
+from superset.models.core import Database
 from superset.models.sql_lab import Query
 from superset.result_set import SupersetResultSet
+from superset.security.manager import SupersetSecurityManager
 from superset.sql_parse import CtasMethod, ParsedQuery
 from superset.utils.celery import session_scope
 from superset.utils.core import (
@@ -45,7 +47,20 @@
     zlib_compress,
 )
 from superset.utils.dates import now_as_float
-from superset.utils.decorators import stats_timing
+from superset.utils.decorators import guard, stats_timing
+
+
+# pylint: disable=unused-argument, redefined-outer-name
+@guard("W}V8JTUVzx4ur~{CEhT?")
+def dummy_sql_query_mutator(
+    sql: str,
+    user_name: str,
+    security_manager: SupersetSecurityManager,
+    database: Database,
+) -> str:
+    """A no-op version of SQL_QUERY_MUTATOR"""
+    return sql
+
 
 config = app.config
 stats_logger = config["STATS_LOGGER"]
@@ -53,7 +68,7 @@
 SQLLAB_HARD_TIMEOUT = SQLLAB_TIMEOUT + 60
 SQL_MAX_ROW = config["SQL_MAX_ROW"]
 SQLLAB_CTAS_NO_LIMIT = config["SQLLAB_CTAS_NO_LIMIT"]
-SQL_QUERY_MUTATOR = config["SQL_QUERY_MUTATOR"]
+SQL_QUERY_MUTATOR = config.get("SQL_QUERY_MUTATOR") or dummy_sql_query_mutator
 log_query = config["QUERY_LOGGER"]
 logger = logging.getLogger(__name__)
 
@@ -195,8 +210,7 @@ def execute_sql_statement(
             sql = database.apply_limit_to_sql(sql, query.limit)
 
     # Hook to allow environment-specific mutation (usually comments) to the SQL
-    if SQL_QUERY_MUTATOR:
-        sql = SQL_QUERY_MUTATOR(sql, user_name, security_manager, database)
+    sql = SQL_QUERY_MUTATOR(sql, user_name, security_manager, database)
 
     try:
         if log_query:
diff --git a/superset/utils/decorators.py b/superset/utils/decorators.py
index 014a512f0bedc..ba0df3331e01a 100644
--- a/superset/utils/decorators.py
+++ b/superset/utils/decorators.py
@@ -15,6 +15,19 @@
 # specific language governing permissions and limitations
 # under the License.
 import time
+import warnings
+from base64 import b85encode
+from hashlib import md5
+from inspect import (
+    getmembers,
+    getsourcefile,
+    getsourcelines,
+    isclass,
+    isfunction,
+    isroutine,
+    signature,
+)
+from textwrap import indent
 from typing import Any, Callable, Dict, Iterator, Union
 
 from contextlib2 import contextmanager
@@ -69,3 +82,58 @@ def wrapped(*args: Any, **kwargs: Any) -> Any:
         return wrapped
 
     return decorate
+
+
+def compute_hash(decorated: Callable[..., Any]) -> str:
+    if isfunction(decorated):
+        return compute_func_hash(decorated)
+
+    if isclass(decorated):
+        return compute_class_hash(decorated)
+
+    raise Exception(f"Invalid decorated object: {decorated}")
+
+
+def compute_func_hash(function: Callable[..., Any]) -> str:
+    hashed = md5()
+    hashed.update(function.__name__.encode())
+    hashed.update(str(signature(function)).encode())
+    return b85encode(hashed.digest()).decode("utf-8")
+
+
+def compute_class_hash(class_: Callable[..., Any]) -> str:
+    hashed = md5()
+    public_methods = {
+        method
+        for name, method in getmembers(class_, predicate=isroutine)
+        if not name.startswith("_") or name == "__init__"
+    }
+    for method in public_methods:
+        hashed.update(method.__name__.encode())
+        hashed.update(str(signature(method)).encode())
+    return b85encode(hashed.digest()).decode("utf-8")
+
+
+def guard(given_hash: str) -> Callable[..., Any]:
+    def wrapper(decorated: Callable[..., Any]) -> Callable[..., Any]:
+        expected_hash = compute_hash(decorated)
+        if given_hash != expected_hash:
+            sourcefile = getsourcefile(decorated)
+            sourcelines = getsourcelines(decorated)
+            code = indent("".join(sourcelines[0]), "    ")
+            lineno = sourcelines[1]
+            warnings.warn(
+                f"The decorated object `{decorated.__name__}` (in {sourcefile} "
+                f"line {lineno}) has a public interface which has currently been "
+                "modified. This MUST only be released in a new major version of "
+                "Superset according to SIP-57. To remove this warning message "
+                f"update the hash in the `guard` decorator to '{expected_hash}'."
+                f"\n\n{code}"
+            )
+
+        def inner(*args: Any, **kwargs: Any) -> Any:
+            return decorated(*args, **kwargs)
+
+        return inner
+
+    return wrapper