fix(tags): make SavedQuery delete-listener registration idempotent

Guard sqla.event.listen/remove for the SavedQuery after_delete listener
with sa.event.contains(), matching the pattern register_delete_listener
already uses, so repeated register_sqla_event_listeners() calls don't
stack duplicate cleanup callbacks.

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

UPDATING.md

@@ -26,6 +26,30 @@ assists people when migrating to a new version.
 
 - `SAMPLES_ROW_LIMIT` is now the default for `/datasource/samples` requests without a valid explicit `per_page`, rather than a hard per-request ceiling; explicit limits are honored up to the existing global row-limit ceiling, matching `/chart/data` SAMPLES requests.
 
+### Superset no longer auto-generates `type:`/`editor:`/`favorited_by:` tags
+
+With `TAGGING_SYSTEM` enabled, Superset used to auto-tag every chart,
+dashboard, saved query, and dataset with implicit tags derived from
+metadata (object type, editors, and who favorited it), and generate a
+`favorited_by:<user id>` tag on every favorite/unfavorite. Nothing in the
+UI ever surfaced these tags to users — every tags list and filter in the
+frontend explicitly excluded them — so the generation added continuous
+write overhead (13 SQLAlchemy event listeners across 5 models) with no
+user-visible benefit. That generation is removed.
+
+Manually-created (custom) tags are unaffected: creating, editing,
+listing, and filtering tags still works exactly as before, including the
+`custom_tag` API filter used to distinguish custom from implicit tags.
+
+Deployments already running with `TAGGING_SYSTEM` enabled keep any
+`type:`/`editor:`/`favorited_by:` tag rows created before upgrading — they
+remain queryable via the API and MCP's `list_tags`/`get_tag_info` tools,
+and are still exempt from bulk tag deletion — but no new ones are created,
+and the `superset sync_tags` CLI command that backfilled them has been
+removed. The `DASHBOARD_LIST_CUSTOM_TAGS_ONLY` config flag and the
+dashboard-list optimization it enabled are also removed, since every tag
+returned is now a custom tag by default.
+
 ### MCP tool results preserve stored string values
 
 Structured MCP tool results no longer add `<UNTRUSTED-CONTENT>` wrappers or

docs/static/resources/openapi.json

@@ -3257,12 +3257,6 @@
             "description": "Override CSS for the dashboard.",
             "type": "string"
           },
-          "custom_tags": {
-            "items": {
-              "$ref": "#/components/schemas/Tag1"
-            },
-            "type": "array"
-          },
           "dashboard_title": {
             "description": "A title for the dashboard.",
             "type": "string"
@@ -17207,7 +17201,6 @@
                     "charts": [],
                     "created_on_delta_humanized": "string",
                     "css": "string",
-                    "custom_tags": [],
                     "dashboard_title": "string",
                     "id": 1,
                     "is_managed_externally": true,

superset/app.py

@@ -194,7 +194,8 @@ class SupersetApp(Flask):
 
             logger.info("Syncing configuration to database...")
 
-            # Register SQLA event listeners for tagging system
+            # Register the tagged_object cleanup listeners for the tagging system
+            # (deletion only; see superset.tags.core.register_sqla_event_listeners)
             if feature_flag_manager.is_feature_enabled("TAGGING_SYSTEM"):
                 from superset.tags.core import register_sqla_event_listeners
 

superset/cli/update.py

@@ -25,7 +25,6 @@ from apispec import APISpec
 from apispec.ext.marshmallow import MarshmallowPlugin
 from flask import current_app
 from flask.cli import with_appcontext
-from flask_appbuilder import Model
 from flask_appbuilder.api import BaseApi
 from flask_appbuilder.api.manager import resolver
 
@@ -53,22 +52,6 @@ def set_database_uri(database_name: str, uri: str, skip_create: bool) -> None:
     database_utils.get_or_create_db(database_name, uri, not skip_create)
 
 
-@click.command()
-@with_appcontext
-@transaction()
-def sync_tags() -> None:
-    """Rebuilds special tags (owner, type, favorited by)."""
-    # pylint: disable=no-member
-    metadata = Model.metadata
-
-    # pylint: disable=import-outside-toplevel
-    from superset.common.tags import add_favorites, add_owners, add_types
-
-    add_types(metadata)
-    add_owners(metadata)
-    add_favorites(metadata)
-
-
 @click.command()
 @with_appcontext
 def update_api_docs() -> None:

superset/common/tags.py

@@ -1,496 +0,0 @@
-# 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 contextlib
-from typing import Any
-
-from sqlalchemy import MetaData
-from sqlalchemy.exc import IntegrityError
-from sqlalchemy.sql import and_, func, join, literal, select
-
-from superset.extensions import db
-from superset.tags.models import ObjectType, TagType
-
-
-def add_types_to_charts(
-    metadata: MetaData, tag: Any, tagged_object: Any, columns: list[str]
-) -> None:
-    slices = metadata.tables["slices"]
-
-    charts = (
-        select(
-            tag.c.id.label("tag_id"),
-            slices.c.id.label("object_id"),
-            literal(ObjectType.chart.name).label("object_type"),
-        )
-        .select_from(
-            join(
-                join(slices, tag, tag.c.name == "type:chart"),
-                tagged_object,
-                and_(
-                    tagged_object.c.tag_id == tag.c.id,
-                    tagged_object.c.object_id == slices.c.id,
-                    tagged_object.c.object_type == "chart",
-                ),
-                isouter=True,
-                full=False,
-            )
-        )
-        .where(tagged_object.c.tag_id.is_(None))
-    )
-    query = tagged_object.insert().from_select(columns, charts)
-    db.session.execute(query)
-
-
-def add_types_to_dashboards(
-    metadata: MetaData, tag: Any, tagged_object: Any, columns: list[str]
-) -> None:
-    dashboard_table = metadata.tables["dashboards"]
-
-    dashboards = (
-        select(
-            tag.c.id.label("tag_id"),
-            dashboard_table.c.id.label("object_id"),
-            literal(ObjectType.dashboard.name).label("object_type"),
-        )
-        .select_from(
-            join(
-                join(dashboard_table, tag, tag.c.name == "type:dashboard"),
-                tagged_object,
-                and_(
-                    tagged_object.c.tag_id == tag.c.id,
-                    tagged_object.c.object_id == dashboard_table.c.id,
-                    tagged_object.c.object_type == "dashboard",
-                ),
-                isouter=True,
-                full=False,
-            )
-        )
-        .where(tagged_object.c.tag_id.is_(None))
-    )
-    query = tagged_object.insert().from_select(columns, dashboards)
-    db.session.execute(query)
-
-
-def add_types_to_saved_queries(
-    metadata: MetaData, tag: Any, tagged_object: Any, columns: list[str]
-) -> None:
-    saved_query = metadata.tables["saved_query"]
-
-    saved_queries = (
-        select(
-            tag.c.id.label("tag_id"),
-            saved_query.c.id.label("object_id"),
-            literal(ObjectType.query.name).label("object_type"),
-        )
-        .select_from(
-            join(
-                join(saved_query, tag, tag.c.name == "type:query"),
-                tagged_object,
-                and_(
-                    tagged_object.c.tag_id == tag.c.id,
-                    tagged_object.c.object_id == saved_query.c.id,
-                    tagged_object.c.object_type == "query",
-                ),
-                isouter=True,
-                full=False,
-            )
-        )
-        .where(tagged_object.c.tag_id.is_(None))
-    )
-    query = tagged_object.insert().from_select(columns, saved_queries)
-    db.session.execute(query)
-
-
-def add_types_to_datasets(
-    metadata: MetaData, tag: Any, tagged_object: Any, columns: list[str]
-) -> None:
-    tables = metadata.tables["tables"]
-
-    datasets = (
-        select(
-            tag.c.id.label("tag_id"),
-            tables.c.id.label("object_id"),
-            literal(ObjectType.dataset.name).label("object_type"),
-        )
-        .select_from(
-            join(
-                join(tables, tag, tag.c.name == "type:dataset"),
-                tagged_object,
-                and_(
-                    tagged_object.c.tag_id == tag.c.id,
-                    tagged_object.c.object_id == tables.c.id,
-                    tagged_object.c.object_type == "dataset",
-                ),
-                isouter=True,
-                full=False,
-            )
-        )
-        .where(tagged_object.c.tag_id.is_(None))
-    )
-    query = tagged_object.insert().from_select(columns, datasets)
-    db.session.execute(query)
-
-
-def add_types(metadata: MetaData) -> None:
-    """
-    Tag every object according to its type:
-
-      INSERT INTO tagged_object (tag_id, object_id, object_type)
-      SELECT
-        tag.id AS tag_id,
-        slices.id AS object_id,
-        'chart' AS object_type
-      FROM slices
-      JOIN tag
-        ON tag.name = 'type:chart'
-      LEFT OUTER JOIN tagged_object
-        ON tagged_object.tag_id = tag.id
-        AND tagged_object.object_id = slices.id
-        AND tagged_object.object_type = 'chart'
-      WHERE tagged_object.tag_id IS NULL;
-
-      INSERT INTO tagged_object (tag_id, object_id, object_type)
-      SELECT
-        tag.id AS tag_id,
-        dashboards.id AS object_id,
-        'dashboard' AS object_type
-      FROM dashboards
-      JOIN tag
-      ON tag.name = 'type:dashboard'
-      LEFT OUTER JOIN tagged_object
-        ON tagged_object.tag_id = tag.id
-        AND tagged_object.object_id = dashboards.id
-        AND tagged_object.object_type = 'dashboard'
-      WHERE tagged_object.tag_id IS NULL;
-
-      INSERT INTO tagged_object (tag_id, object_id, object_type)
-      SELECT
-        tag.id AS tag_id,
-        saved_query.id AS object_id,
-        'query' AS object_type
-      FROM saved_query
-      JOIN tag
-      ON tag.name = 'type:query';
-      LEFT OUTER JOIN tagged_object
-        ON tagged_object.tag_id = tag.id
-        AND tagged_object.object_id = saved_query.id
-        AND tagged_object.object_type = 'query'
-      WHERE tagged_object.tag_id IS NULL;
-
-      INSERT INTO tagged_object (tag_id, object_id, object_type)
-      SELECT
-        tag.id AS tag_id,
-        tables.id AS object_id,
-        'dataset' AS object_type
-      FROM tables
-      JOIN tag
-        ON tag.name = 'type:dataset'
-      LEFT OUTER JOIN tagged_object
-        ON tagged_object.tag_id = tag.id
-        AND tagged_object.object_id = tables.id
-        AND tagged_object.object_type = 'dataset'
-      WHERE tagged_object.tag_id IS NULL;
-
-    """
-
-    tag = metadata.tables["tag"]
-    tagged_object = metadata.tables["tagged_object"]
-    columns = ["tag_id", "object_id", "object_type"]
-
-    # add a tag for each object type
-    insert = tag.insert()
-    for type_ in ObjectType.__members__:
-        with contextlib.suppress(IntegrityError):  # already exists
-            db.session.execute(insert, name=f"type:{type_}", type=TagType.type)
-
-    add_types_to_charts(metadata, tag, tagged_object, columns)
-    add_types_to_dashboards(metadata, tag, tagged_object, columns)
-    add_types_to_saved_queries(metadata, tag, tagged_object, columns)
-    add_types_to_datasets(metadata, tag, tagged_object, columns)
-
-
-def add_owners_to_charts(
-    metadata: MetaData, tag: Any, tagged_object: Any, columns: list[str]
-) -> None:
-    slices = metadata.tables["slices"]
-
-    charts = (
-        select(
-            tag.c.id.label("tag_id"),
-            slices.c.id.label("object_id"),
-            literal(ObjectType.chart.name).label("object_type"),
-        )
-        .select_from(
-            join(
-                join(
-                    slices,
-                    tag,
-                    tag.c.name == "editor:" + slices.c.created_by_fk,
-                ),
-                tagged_object,
-                and_(
-                    tagged_object.c.tag_id == tag.c.id,
-                    tagged_object.c.object_id == slices.c.id,
-                    tagged_object.c.object_type == "chart",
-                ),
-                isouter=True,
-                full=False,
-            )
-        )
-        .where(tagged_object.c.tag_id.is_(None))
-    )
-    query = tagged_object.insert().from_select(columns, charts)
-    db.session.execute(query)
-
-
-def add_owners_to_dashboards(
-    metadata: MetaData, tag: Any, tagged_object: Any, columns: list[str]
-) -> None:
-    dashboard_table = metadata.tables["dashboards"]
-
-    dashboards = (
-        select(
-            tag.c.id.label("tag_id"),
-            dashboard_table.c.id.label("object_id"),
-            literal(ObjectType.dashboard.name).label("object_type"),
-        )
-        .select_from(
-            join(
-                join(
-                    dashboard_table,
-                    tag,
-                    tag.c.name == "editor:" + dashboard_table.c.created_by_fk,
-                ),
-                tagged_object,
-                and_(
-                    tagged_object.c.tag_id == tag.c.id,
-                    tagged_object.c.object_id == dashboard_table.c.id,
-                    tagged_object.c.object_type == "dashboard",
-                ),
-                isouter=True,
-                full=False,
-            )
-        )
-        .where(tagged_object.c.tag_id.is_(None))
-    )
-    query = tagged_object.insert().from_select(columns, dashboards)
-    db.session.execute(query)
-
-
-def add_owners_to_saved_queries(
-    metadata: MetaData, tag: Any, tagged_object: Any, columns: list[str]
-) -> None:
-    saved_query = metadata.tables["saved_query"]
-
-    saved_queries = (
-        select(
-            tag.c.id.label("tag_id"),
-            saved_query.c.id.label("object_id"),
-            literal(ObjectType.query.name).label("object_type"),
-        )
-        .select_from(
-            join(
-                join(
-                    saved_query,
-                    tag,
-                    tag.c.name == "editor:" + saved_query.c.created_by_fk,
-                ),
-                tagged_object,
-                and_(
-                    tagged_object.c.tag_id == tag.c.id,
-                    tagged_object.c.object_id == saved_query.c.id,
-                    tagged_object.c.object_type == "query",
-                ),
-                isouter=True,
-                full=False,
-            )
-        )
-        .where(tagged_object.c.tag_id.is_(None))
-    )
-    query = tagged_object.insert().from_select(columns, saved_queries)
-    db.session.execute(query)
-
-
-def add_owners_to_datasets(
-    metadata: MetaData, tag: Any, tagged_object: Any, columns: list[str]
-) -> None:
-    tables = metadata.tables["tables"]
-
-    datasets = (
-        select(
-            tag.c.id.label("tag_id"),
-            tables.c.id.label("object_id"),
-            literal(ObjectType.dataset.name).label("object_type"),
-        )
-        .select_from(
-            join(
-                join(
-                    tables,
-                    tag,
-                    tag.c.name == "editor:" + tables.c.created_by_fk,
-                ),
-                tagged_object,
-                and_(
-                    tagged_object.c.tag_id == tag.c.id,
-                    tagged_object.c.object_id == tables.c.id,
-                    tagged_object.c.object_type == "dataset",
-                ),
-                isouter=True,
-                full=False,
-            )
-        )
-        .where(tagged_object.c.tag_id.is_(None))
-    )
-    query = tagged_object.insert().from_select(columns, datasets)
-    db.session.execute(query)
-
-
-def add_owners(metadata: MetaData) -> None:
-    """
-    Tag every object according to its editor:
-
-      INSERT INTO tagged_object (tag_id, object_id, object_type)
-      SELECT
-        tag.id AS tag_id,
-        slices.id AS object_id,
-        'chart' AS object_type
-      FROM slices
-      JOIN tag
-      ON tag.name = CONCAT('editor:', slices.created_by_fk)
-      LEFT OUTER JOIN tagged_object
-        ON tagged_object.tag_id = tag.id
-        AND tagged_object.object_id = slices.id
-        AND tagged_object.object_type = 'chart'
-      WHERE tagged_object.tag_id IS NULL;
-
-      SELECT
-        tag.id AS tag_id,
-        dashboards.id AS object_id,
-        'dashboard' AS object_type
-      FROM dashboards
-      JOIN tag
-      ON tag.name = CONCAT('editor:', dashboards.created_by_fk)
-      LEFT OUTER JOIN tagged_object
-        ON tagged_object.tag_id = tag.id
-        AND tagged_object.object_id = dashboards.id
-        AND tagged_object.object_type = 'dashboard'
-      WHERE tagged_object.tag_id IS NULL;
-
-      SELECT
-        tag.id AS tag_id,
-        saved_query.id AS object_id,
-        'query' AS object_type
-      FROM saved_query
-      JOIN tag
-      ON tag.name = CONCAT('editor:', saved_query.created_by_fk)
-      LEFT OUTER JOIN tagged_object
-        ON tagged_object.tag_id = tag.id
-        AND tagged_object.object_id = saved_query.id
-        AND tagged_object.object_type = 'query'
-      WHERE tagged_object.tag_id IS NULL;
-
-      SELECT
-        tag.id AS tag_id,
-        tables.id AS object_id,
-        'dataset' AS object_type
-      FROM tables
-      JOIN tag
-      ON tag.name = CONCAT('editor:', tables.created_by_fk)
-      LEFT OUTER JOIN tagged_object
-        ON tagged_object.tag_id = tag.id
-        AND tagged_object.object_id = tables.id
-        AND tagged_object.object_type = 'dataset'
-      WHERE tagged_object.tag_id IS NULL;
-
-    """
-
-    tag = metadata.tables["tag"]
-    tagged_object = metadata.tables["tagged_object"]
-    users = metadata.tables["ab_user"]
-    columns = ["tag_id", "object_id", "object_type"]
-
-    # create a custom tag for each user
-    ids = select(users.c.id)
-    insert = tag.insert()
-    for (id_,) in db.session.execute(ids):
-        with contextlib.suppress(IntegrityError):  # already exists
-            db.session.execute(insert, name=f"editor:{id_}", type=TagType.editor)
-    add_owners_to_charts(metadata, tag, tagged_object, columns)
-    add_owners_to_dashboards(metadata, tag, tagged_object, columns)
-    add_owners_to_saved_queries(metadata, tag, tagged_object, columns)
-    add_owners_to_datasets(metadata, tag, tagged_object, columns)
-
-
-def add_favorites(metadata: MetaData) -> None:
-    """
-    Tag every object that was favorited:
-
-      INSERT INTO tagged_object (tag_id, object_id, object_type)
-      SELECT
-        tag.id AS tag_id,
-        favstar.obj_id AS object_id,
-        LOWER(favstar.class_name) AS object_type
-      FROM favstar
-      JOIN tag
-      ON tag.name = CONCAT('favorited_by:', favstar.user_id)
-      LEFT OUTER JOIN tagged_object
-        ON tagged_object.tag_id = tag.id
-        AND tagged_object.object_id = favstar.obj_id
-        AND tagged_object.object_type = LOWER(favstar.class_name)
-      WHERE tagged_object.tag_id IS NULL;
-
-    """
-
-    tag = metadata.tables["tag"]
-    tagged_object = metadata.tables["tagged_object"]
-    users = metadata.tables["ab_user"]
-    favstar = metadata.tables["favstar"]
-    columns = ["tag_id", "object_id", "object_type"]
-
-    # create a custom tag for each user
-    ids = select(users.c.id)
-    insert = tag.insert()
-    for (id_,) in db.session.execute(ids):
-        with contextlib.suppress(IntegrityError):  # already exists
-            db.session.execute(insert, name=f"favorited_by:{id_}", type=TagType.type)
-    favstars = (
-        select(
-            tag.c.id.label("tag_id"),
-            favstar.c.obj_id.label("object_id"),
-            func.lower(favstar.c.class_name).label("object_type"),
-        )
-        .select_from(
-            join(
-                join(
-                    favstar,
-                    tag,
-                    tag.c.name == "favorited_by:" + favstar.c.user_id,
-                ),
-                tagged_object,
-                and_(
-                    tagged_object.c.tag_id == tag.c.id,
-                    tagged_object.c.object_id == favstar.c.obj_id,
-                    tagged_object.c.object_type == func.lower(favstar.c.class_name),
-                ),
-                isouter=True,
-                full=False,
-            )
-        )
-        .where(tagged_object.c.tag_id.is_(None))
-    )
-    query = tagged_object.insert().from_select(columns, favstars)
-    db.session.execute(query)

superset/config.py

@@ -1782,11 +1782,6 @@ DASHBOARD_AUTO_REFRESH_INTERVALS = [
     [86400, "24 hours"],
 ]
 
-# Performance optimization: Return only custom tags in dashboard list API
-# When enabled, filters out implicit tags (owner, type, favorited_by) at SQL JOIN level
-# Reduces response payload and query time for dashboards with many editors
-DASHBOARD_LIST_CUSTOM_TAGS_ONLY: bool = False
-
 # This is used as a workaround for the alerts & reports scheduler task to get the time
 # celery beat triggered it, see https://github.com/celery/celery/issues/6974 for details
 CELERY_BEAT_SCHEDULER_EXPIRES = timedelta(weeks=1)

superset/dashboards/api.py

@@ -188,7 +188,6 @@ from superset.views.base_api import (
     statsd_metrics,
     validate_feature_flags,
 )
-from superset.views.custom_tags_api_mixin import CustomTagsOptimizationMixin
 from superset.views.error_handling import handle_api_exception
 from superset.views.filters import (
     BaseFilterRelatedUsers,
@@ -259,20 +258,12 @@ BASE_LIST_COLUMNS = [
     "uuid",
 ]
 
-# Full tags (current behavior - includes all tag types)
-FULL_TAG_LIST_COLUMNS = BASE_LIST_COLUMNS + [
+TAG_LIST_COLUMNS = BASE_LIST_COLUMNS + [
     "tags.id",
     "tags.name",
     "tags.type",
 ]
 
-# Custom tags only
-CUSTOM_TAG_LIST_COLUMNS = BASE_LIST_COLUMNS + [
-    "custom_tags.id",
-    "custom_tags.name",
-    "custom_tags.type",
-]
-
 # Fields dropped from a dashboard member dataset when the caller cannot access
 # that datasource on its own: everything describing the dataset's schema,
 # connection, and query construction. The identifying fields the dashboard
@@ -300,9 +291,7 @@ DASHBOARD_DATASET_INACCESSIBLE_FIELDS = (
 
 
 # pylint: disable=too-many-public-methods
-class DashboardRestApi(
-    SoftDeleteApiMixin, CustomTagsOptimizationMixin, BaseSupersetModelRestApi
-):
+class DashboardRestApi(SoftDeleteApiMixin, BaseSupersetModelRestApi):
     datamodel = SQLAInterface(Dashboard)
 
     include_route_methods = RouteMethod.REST_MODEL_VIEW_CRUD_SET | {
@@ -357,17 +346,7 @@ class DashboardRestApi(
         "purge": "write",
     }
 
-    # Default list_columns (used if config not set)
-    list_columns = FULL_TAG_LIST_COLUMNS
-
-    def __init__(self) -> None:
-        # Configure custom tags optimization (mixin handles the logic)
-        self._setup_custom_tags_optimization(
-            config_key="DASHBOARD_LIST_CUSTOM_TAGS_ONLY",
-            full_columns=FULL_TAG_LIST_COLUMNS,
-            custom_columns=CUSTOM_TAG_LIST_COLUMNS,
-        )
-        super().__init__()
+    list_columns = TAG_LIST_COLUMNS
 
     @expose("/", methods=("GET",))
     @protect()

superset/dashboards/schemas.py

@@ -292,7 +292,6 @@ class DashboardGetResponseSchema(Schema):
     editors = fields.List(fields.Nested(SubjectResponseSchema))
     viewers = fields.List(fields.Nested(SubjectResponseSchema))
     tags = fields.Nested(TagSchema, many=True)
-    custom_tags = fields.Nested(TagSchema, many=True)
     changed_on_humanized = fields.String(data_key="changed_on_delta_humanized")
     created_on_humanized = fields.String(data_key="created_on_delta_humanized")
     is_managed_externally = fields.Boolean(allow_none=True, dump_default=False)
@@ -302,12 +301,6 @@ class DashboardGetResponseSchema(Schema):
     # pylint: disable=unused-argument
     @post_dump()
     def post_dump(self, serialized: dict[str, Any], **kwargs: Any) -> dict[str, Any]:
-        # Handle custom_tags → tags renaming when flag is enabled
-        # When DASHBOARD_LIST_CUSTOM_TAGS_ONLY=True, FAB populates custom_tags
-        # Rename it to tags for frontend compatibility
-        if "custom_tags" in serialized:
-            serialized["tags"] = serialized.pop("custom_tags")
-
         if security_manager.is_guest_user():
             del serialized["changed_by_name"]
             del serialized["changed_by"]

superset/mcp_service/tag/tool/get_tag_info.py

@@ -54,8 +54,10 @@ async def get_tag_info(request: GetTagInfoRequest, ctx: Context) -> TagInfo | Ta
 
     Returns tag details including name, type, and description.
 
-    Tag types: custom (user-created), type (implicit by object type),
-    editor (implicit by editorship), favorited_by (implicit by favorites).
+    Tag types: custom (user-created), plus legacy implicit types no longer
+    generated for new objects -- type (by object type), editor (by
+    editorship), favorited_by (by favorites) -- kept for tags created before
+    Superset stopped auto-generating them.
 
     To find a tag ID, use the list_tags tool first.
 

superset/mcp_service/tag/tool/list_tags.py

@@ -73,8 +73,10 @@ async def list_tags(
 
     Returns tag metadata including name, type, and description.
 
-    Tag types: custom (user-created), type (implicit by object type),
-    editor (implicit by editorship), favorited_by (implicit by favorites).
+    Tag types: custom (user-created), plus legacy implicit types no longer
+    generated for new objects -- type (by object type), editor (by
+    editorship), favorited_by (by favorites) -- kept for tags created before
+    Superset stopped auto-generating them.
 
     Sortable columns for order_column: id, name, changed_on, created_on
     """

superset/models/dashboard.py

@@ -191,16 +191,6 @@ class Dashboard(CoreDashboard, SoftDeleteMixin, AuditMixinNullable, ImportExport
         secondaryjoin="TaggedObject.tag_id == Tag.id",
         viewonly=True,  # cascading deletion already handled by superset.tags.models.ObjectUpdater.after_delete  # noqa: E501
     )
-    custom_tags = relationship(
-        "Tag",
-        overlaps="objects,tag,tags,custom_tags",
-        secondary="tagged_object",
-        primaryjoin="and_(Dashboard.id == TaggedObject.object_id, "
-        "TaggedObject.object_type == 'dashboard')",
-        secondaryjoin="and_(TaggedObject.tag_id == Tag.id, "
-        "cast(Tag.type, String) == 'custom')",  # Filtering at JOIN level
-        viewonly=True,
-    )
     theme = relationship("Theme", foreign_keys=[theme_id])
     published = Column(Boolean, default=False)
     is_managed_externally = Column(Boolean, nullable=False, default=False)

superset/tags/core.py

@@ -54,80 +54,40 @@ def _tag_delete_listener_declarations() -> tuple[DeleteListenerDeclaration, ...]
 
 
 def register_sqla_event_listeners() -> None:
+    """Register cleanup of ``tagged_object`` rows on object deletion.
+
+    Only deletion is handled here: Superset no longer auto-generates
+    ``type:``/``editor:``/``favorited_by:`` tags (see ``TagType``'s docstring),
+    so there's nothing left to do on insert/update. Deletion cleanup stays,
+    since it applies to every tag on the object, custom tags included, and
+    ``tagged_object.object_id`` has no foreign key to cascade on its own.
+    """
     import sqlalchemy as sqla
 
-    from superset.connectors.sqla.models import SqlaTable
-    from superset.models.core import FavStar
-    from superset.models.dashboard import Dashboard
-    from superset.models.slice import Slice
     from superset.models.sql_lab import SavedQuery
-    from superset.tags.models import (
-        ChartUpdater,
-        DashboardUpdater,
-        DatasetUpdater,
-        FavStarUpdater,
-        QueryUpdater,
-    )
+    from superset.tags.models import QueryUpdater
 
-    declarations: tuple[DeleteListenerDeclaration, ...] = (
-        _tag_delete_listener_declarations()
-    )
+    declarations = _tag_delete_listener_declarations()
 
-    sqla.event.listen(SqlaTable, "after_insert", DatasetUpdater.after_insert)
-    sqla.event.listen(SqlaTable, "after_update", DatasetUpdater.after_update)
-    register_delete_listener(declarations[0])
+    register_delete_listener(declarations[0])  # dataset
+    register_delete_listener(declarations[1])  # chart
+    register_delete_listener(declarations[2])  # dashboard
 
-    sqla.event.listen(Slice, "after_insert", ChartUpdater.after_insert)
-    sqla.event.listen(Slice, "after_update", ChartUpdater.after_update)
-    register_delete_listener(declarations[1])
-
-    sqla.event.listen(Dashboard, "after_insert", DashboardUpdater.after_insert)
-    sqla.event.listen(Dashboard, "after_update", DashboardUpdater.after_update)
-    register_delete_listener(declarations[2])
-
-    sqla.event.listen(FavStar, "after_insert", FavStarUpdater.after_insert)
-    sqla.event.listen(FavStar, "after_delete", FavStarUpdater.after_delete)
-
-    sqla.event.listen(SavedQuery, "after_insert", QueryUpdater.after_insert)
-    sqla.event.listen(SavedQuery, "after_update", QueryUpdater.after_update)
-    sqla.event.listen(SavedQuery, "after_delete", QueryUpdater.after_delete)
+    if not sqla.event.contains(SavedQuery, "after_delete", QueryUpdater.after_delete):
+        sqla.event.listen(SavedQuery, "after_delete", QueryUpdater.after_delete)
 
 
 def clear_sqla_event_listeners() -> None:
     import sqlalchemy as sqla
 
-    from superset.connectors.sqla.models import SqlaTable
-    from superset.models.core import FavStar
-    from superset.models.dashboard import Dashboard
-    from superset.models.slice import Slice
     from superset.models.sql_lab import SavedQuery
-    from superset.tags.models import (
-        ChartUpdater,
-        DashboardUpdater,
-        DatasetUpdater,
-        FavStarUpdater,
-        QueryUpdater,
-    )
+    from superset.tags.models import QueryUpdater
 
-    declarations: tuple[DeleteListenerDeclaration, ...] = (
-        _tag_delete_listener_declarations()
-    )
+    declarations = _tag_delete_listener_declarations()
 
-    sqla.event.remove(SqlaTable, "after_insert", DatasetUpdater.after_insert)
-    sqla.event.remove(SqlaTable, "after_update", DatasetUpdater.after_update)
-    remove_delete_listener(declarations[0])
+    remove_delete_listener(declarations[0])  # dataset
+    remove_delete_listener(declarations[1])  # chart
+    remove_delete_listener(declarations[2])  # dashboard
 
-    sqla.event.remove(Slice, "after_insert", ChartUpdater.after_insert)
-    sqla.event.remove(Slice, "after_update", ChartUpdater.after_update)
-    remove_delete_listener(declarations[1])
-
-    sqla.event.remove(Dashboard, "after_insert", DashboardUpdater.after_insert)
-    sqla.event.remove(Dashboard, "after_update", DashboardUpdater.after_update)
-    remove_delete_listener(declarations[2])
-
-    sqla.event.remove(FavStar, "after_insert", FavStarUpdater.after_insert)
-    sqla.event.remove(FavStar, "after_delete", FavStarUpdater.after_delete)
-
-    sqla.event.remove(SavedQuery, "after_insert", QueryUpdater.after_insert)
-    sqla.event.remove(SavedQuery, "after_update", QueryUpdater.after_update)
-    sqla.event.remove(SavedQuery, "after_delete", QueryUpdater.after_delete)
+    if sqla.event.contains(SavedQuery, "after_delete", QueryUpdater.after_delete):
+        sqla.event.remove(SavedQuery, "after_delete", QueryUpdater.after_delete)

superset/tags/models.py

@@ -21,17 +21,7 @@ import enum
 from typing import TYPE_CHECKING
 
 from flask_appbuilder import Model
-from sqlalchemy import (
-    Column,
-    Enum,
-    exists,
-    ForeignKey,
-    Integer,
-    orm,
-    String,
-    Table,
-    Text,
-)
+from sqlalchemy import Column, Enum, ForeignKey, Integer, orm, String, Table, Text
 from sqlalchemy.engine.base import Connection
 from sqlalchemy.orm import relationship, sessionmaker
 from sqlalchemy.orm.mapper import Mapper
@@ -40,11 +30,9 @@ from superset_core.common.models import Tag as CoreTag
 
 from superset import security_manager
 from superset.models.helpers import AuditMixinNullable
-from superset.subjects.types import SubjectType
 
 if TYPE_CHECKING:
     from superset.connectors.sqla.models import SqlaTable
-    from superset.models.core import FavStar
     from superset.models.dashboard import Dashboard
     from superset.models.slice import Slice
     from superset.models.sql_lab import Query
@@ -63,16 +51,21 @@ class TagType(enum.Enum):
     """
     Types for tags.
 
-    Objects (queries, charts, dashboards, and datasets) will have with implicit tags based
-    on metadata: types, editors and who favorited them. This way, user "alice"
-    can find all their objects by querying for the tag `editor:alice`.
-    """  # noqa: E501
+    ``type``, ``editor``, and ``favorited_by`` are no longer generated: Superset
+    used to auto-tag every query, chart, dashboard, and dataset with implicit
+    tags based on metadata (object type, editors, and who favorited them), but
+    nothing ever surfaced them to the user, so the generation was removed. The
+    values are kept, and rows of these types are still recognized (e.g. exempt
+    from bulk deletion) and filterable via the API, so upgraded deployments that
+    already have such tags, or MCP tooling that queries by tag type, keep
+    working.
+    """
 
     # pylint: disable=invalid-name
     # explicit tags, added manually by the owner
     custom = 1
 
-    # implicit tags, generated automatically
+    # legacy implicit tag types; no longer generated (see docstring above)
     type = 2
     editor = 3
     favorited_by = 4
@@ -155,145 +148,26 @@ def get_tag(
     return tag
 
 
-def get_object_type(class_name: str) -> ObjectType:
-    mapping = {
-        "slice": ObjectType.chart,
-        "dashboard": ObjectType.dashboard,
-        "query": ObjectType.query,
-        "dataset": ObjectType.dataset,
-    }
-    try:
-        return mapping[class_name.lower()]
-    except KeyError as ex:
-        raise Exception(  # pylint: disable=broad-exception-raised
-            f"No mapping found for {class_name}"
-        ) from ex
-
-
 class ObjectUpdater:
+    """Cleans up ``tagged_object`` rows when a tagged object is deleted.
+
+    ``TaggedObject.object_id`` is a polymorphic reference with no foreign key
+    (see the comment on that column), so nothing at the database level removes
+    a tag association when the dashboard/chart/query/dataset it points at is
+    deleted. This listener does that cleanup for every tag on the object
+    (custom tags included), independent of how the tag was created.
+    """
+
     object_type: str = "default"
 
-    @classmethod
-    def get_editor_user_ids(
-        cls, target: Dashboard | FavStar | Slice | Query | SqlaTable
-    ) -> list[int]:
-        raise NotImplementedError("Subclass should implement `get_editor_user_ids`")
-
-    @classmethod
-    def get_editor_tag_ids(
-        cls,
-        session: orm.Session,  # pylint: disable=disallowed-name
-        target: Dashboard | FavStar | Slice | Query | SqlaTable,
-    ) -> set[int]:
-        tag_ids = set()
-        for user_id in cls.get_editor_user_ids(target):
-            name = f"editor:{user_id}"
-            tag = get_tag(name, session, TagType.editor)
-            tag_ids.add(tag.id)
-        return tag_ids
-
-    @classmethod
-    def _add_editors(
-        cls,
-        session: orm.Session,  # pylint: disable=disallowed-name
-        target: Dashboard | FavStar | Slice | Query | SqlaTable,
-    ) -> None:
-        for user_id in cls.get_editor_user_ids(target):
-            name: str = f"editor:{user_id}"
-            tag = get_tag(name, session, TagType.editor)
-            cls.add_tag_object_if_not_tagged(
-                session, tag_id=tag.id, object_id=target.id, object_type=cls.object_type
-            )
-
-    @classmethod
-    def add_tag_object_if_not_tagged(
-        cls,
-        session: orm.Session,  # pylint: disable=disallowed-name
-        tag_id: int,
-        object_id: int,
-        object_type: str,
-    ) -> None:
-        # Check if the object is already tagged
-        exists_query = exists().where(
-            TaggedObject.tag_id == tag_id,
-            TaggedObject.object_id == object_id,
-            TaggedObject.object_type == object_type,
-        )
-        already_tagged = session.query(exists_query).scalar()
-
-        # Add TaggedObject to the session if it isn't already tagged
-        if not already_tagged:
-            tagged_object = TaggedObject(
-                tag_id=tag_id, object_id=object_id, object_type=object_type
-            )
-            session.add(tagged_object)
-
-    @classmethod
-    def after_insert(
-        cls,
-        _mapper: Mapper,
-        connection: Connection,
-        target: Dashboard | FavStar | Slice | Query | SqlaTable,
-    ) -> None:
-        with Session(bind=connection) as session:  # pylint: disable=disallowed-name
-            # add `editor:` tags
-            cls._add_editors(session, target)
-
-            # add `type:` tags
-            tag = get_tag(f"type:{cls.object_type}", session, TagType.type)
-            cls.add_tag_object_if_not_tagged(
-                session, tag_id=tag.id, object_id=target.id, object_type=cls.object_type
-            )
-            session.commit()
-
-    @classmethod
-    def after_update(
-        cls,
-        _mapper: Mapper,
-        connection: Connection,
-        target: Dashboard | FavStar | Slice | Query | SqlaTable,
-    ) -> None:
-        with Session(bind=connection) as session:  # pylint: disable=disallowed-name
-            # Fetch current editor tags
-            existing_tags = (
-                session.query(TaggedObject)
-                .join(Tag)
-                .filter(
-                    TaggedObject.object_type == cls.object_type,
-                    TaggedObject.object_id == target.id,
-                    Tag.type == TagType.editor,
-                )
-                .all()
-            )
-            existing_editor_tag_ids = {tag.tag_id for tag in existing_tags}
-
-            # Determine new editor IDs
-            new_editor_tag_ids = cls.get_editor_tag_ids(session, target)
-
-            # Add missing tags
-            for editor_tag_id in new_editor_tag_ids - existing_editor_tag_ids:
-                tagged_object = TaggedObject(
-                    tag_id=editor_tag_id,
-                    object_id=target.id,
-                    object_type=cls.object_type,
-                )
-                session.add(tagged_object)
-
-            # Remove unnecessary tags
-            for tag in existing_tags:
-                if tag.tag_id not in new_editor_tag_ids:
-                    session.delete(tag)
-            session.commit()
-
     @classmethod
     def after_delete(
         cls,
         _mapper: Mapper,
         connection: Connection,
-        target: Dashboard | FavStar | Slice | Query | SqlaTable,
+        target: Dashboard | Slice | Query | SqlaTable,
     ) -> None:
         with Session(bind=connection) as session:  # pylint: disable=disallowed-name
-            # delete row from `tagged_objects`
             session.query(TaggedObject).filter(
                 TaggedObject.object_type == cls.object_type,
                 TaggedObject.object_id == target.id,
@@ -305,75 +179,14 @@ class ObjectUpdater:
 class ChartUpdater(ObjectUpdater):
     object_type = "chart"
 
-    @classmethod
-    def get_editor_user_ids(cls, target: Slice) -> list[int]:
-        return [
-            s.user.id for s in target.editors if s.type == SubjectType.USER and s.user
-        ]
-
 
 class DashboardUpdater(ObjectUpdater):
     object_type = "dashboard"
 
-    @classmethod
-    def get_editor_user_ids(cls, target: Dashboard) -> list[int]:
-        return [
-            s.user.id for s in target.editors if s.type == SubjectType.USER and s.user
-        ]
-
 
 class QueryUpdater(ObjectUpdater):
     object_type = "query"
 
-    @classmethod
-    def get_editor_user_ids(cls, target: Query) -> list[int]:
-        return [target.user_id]
-
 
 class DatasetUpdater(ObjectUpdater):
     object_type = "dataset"
-
-    @classmethod
-    def get_editor_user_ids(cls, target: SqlaTable) -> list[int]:
-        return [
-            s.user.id for s in target.editors if s.type == SubjectType.USER and s.user
-        ]
-
-
-class FavStarUpdater:
-    @classmethod
-    def after_insert(
-        cls, _mapper: Mapper, connection: Connection, target: FavStar
-    ) -> None:
-        with Session(bind=connection) as session:  # pylint: disable=disallowed-name
-            name = f"favorited_by:{target.user_id}"
-            tag = get_tag(name, session, TagType.favorited_by)
-            tagged_object = TaggedObject(
-                tag_id=tag.id,
-                object_id=target.obj_id,
-                object_type=get_object_type(target.class_name),
-            )
-            session.add(tagged_object)
-            session.commit()
-
-    @classmethod
-    def after_delete(
-        cls, _mapper: Mapper, connection: Connection, target: FavStar
-    ) -> None:
-        with Session(bind=connection) as session:  # pylint: disable=disallowed-name
-            name = f"favorited_by:{target.user_id}"
-            query = (
-                session.query(TaggedObject.id)
-                .join(Tag)
-                .filter(
-                    TaggedObject.object_id == target.obj_id,
-                    Tag.type == TagType.favorited_by,
-                    Tag.name == name,
-                )
-            )
-            ids = [row[0] for row in query]
-            session.query(TaggedObject).filter(TaggedObject.id.in_(ids)).delete(
-                synchronize_session=False
-            )
-
-            session.commit()

superset/views/custom_tags_api_mixin.py

@@ -1,118 +0,0 @@
-# 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.
-"""Mixin for APIs that need custom_tags optimization with frontend compatibility."""
-
-from typing import Any
-
-from flask import current_app, request, Response
-from werkzeug.datastructures import ImmutableMultiDict
-
-
-class CustomTagsOptimizationMixin:
-    """Reusable mixin for APIs that optimize tag queries via custom_tags relationship.
-
-    When enabled via config, this mixin:
-    1. Configures list_columns to use custom_tags (filtered relationship)
-    2. Exposes custom_tags as tags in the response schema
-    3. Rewrites frontend requests from 'tags.*' to 'custom_tags.*'
-    4. Transforms responses to rename 'custom_tags' back to 'tags'
-
-    This provides SQL query optimization (97% reduction) while maintaining
-    frontend compatibility.
-
-    Usage:
-        class MyRestApi(CustomTagsOptimizationMixin, BaseSupersetModelRestApi):
-            def __init__(self):
-                self._setup_custom_tags_optimization(
-                    config_key="MY_API_CUSTOM_TAGS_ONLY",
-                    full_columns=FULL_TAG_COLUMNS,
-                    custom_columns=CUSTOM_TAG_COLUMNS,
-                )
-                super().__init__()
-    """
-
-    _custom_tags_only: bool
-
-    def _setup_custom_tags_optimization(
-        self,
-        config_key: str,
-        full_columns: list[str],
-        custom_columns: list[str],
-    ) -> None:
-        """Configure custom tags optimization based on config.
-
-        Args:
-            config_key: Config key to check (e.g., "DASHBOARD_LIST_CUSTOM_TAGS_ONLY")
-            full_columns: list_columns when optimization disabled (includes all tags)
-            custom_columns: list_columns when optimization enabled (only custom_tags)
-        """
-        self._custom_tags_only = current_app.config.get(config_key, False)
-        self.list_columns = custom_columns if self._custom_tags_only else full_columns
-
-    def _init_model_schemas(self) -> None:
-        """Keep the optimized relationship's public schema name stable."""
-        super()._init_model_schemas()  # type: ignore[misc]
-
-        list_model_schema = getattr(self, "list_model_schema", None)
-        if (
-            self._custom_tags_only
-            and list_model_schema
-            and "custom_tags" in list_model_schema.fields
-        ):
-            list_model_schema.fields["custom_tags"].data_key = "tags"
-
-    def get_list(self, **kwargs: Any) -> Response:
-        """Override to rewrite request parameters for custom_tags optimization.
-
-        When config is enabled, rewrites 'tags.*' → 'custom_tags.*' in request
-        so FAB can find the columns in list_columns.
-        """
-        if self._custom_tags_only:
-            # Parse and rewrite query parameter
-            query_str = request.args.get("q", "")
-            if query_str and "tags." in query_str:
-                # Replace 'tags.' with 'custom_tags.' in select_columns
-                modified_query = query_str.replace("tags.id", "custom_tags.id")
-                modified_query = modified_query.replace("tags.name", "custom_tags.name")
-                modified_query = modified_query.replace("tags.type", "custom_tags.type")
-
-                # Temporarily patch request.args
-                modified_args = request.args.copy()
-                modified_args["q"] = modified_query
-                original_args = request.args
-                request.args = ImmutableMultiDict(modified_args)
-
-                try:
-                    return super().get_list(**kwargs)  # type: ignore
-                finally:
-                    # Restore original args
-                    request.args = original_args
-
-        return super().get_list(**kwargs)  # type: ignore
-
-    def pre_get_list(self, data: dict[str, Any]) -> None:
-        """Rename custom_tags → tags in response for frontend compatibility.
-
-        Called by FAB before sending the list response. This ensures the frontend
-        always receives 'tags' regardless of backend optimization config.
-        """
-        if self._custom_tags_only and "result" in data:
-            for item in data["result"]:
-                if "custom_tags" in item:
-                    item["tags"] = item.pop("custom_tags")
-
-        super().pre_get_list(data)  # type: ignore

tests/integration_tests/dashboards/api_tests.py

@@ -4623,107 +4623,3 @@ class TestDashboardApi(ApiEditorsTestCaseMixin, InsertChartMixin, SupersetTestCa
             # Cleanup
             db.session.delete(dashboard)
             db.session.commit()
-
-
-class TestDashboardCustomTagsFiltering(SupersetTestCase):
-    """Test dashboard list API tags field behavior.
-
-    Note: DASHBOARD_LIST_CUSTOM_TAGS_ONLY config is checked at app startup in
-    DashboardRestApi.__init__(), so these tests verify the current runtime behavior.
-    """
-
-    def setUp(self) -> None:
-        """Set up test fixtures."""
-        self.login(username="admin")
-
-    @pytest.mark.usefixtures("with_tagging_system_feature")
-    def test_dashboard_custom_tags_relationship_filters_correctly(self):
-        """Verify custom_tags filtering at model and API level.
-
-        With DASHBOARD_LIST_CUSTOM_TAGS_ONLY=True in superset_test_config.py:
-        1. dashboard.tags returns ALL tags (custom + editor + type)
-        2. dashboard.custom_tags returns ONLY custom tags
-        3. API response returns ONLY custom tags in the "tags" property
-        """
-        dashboard = Dashboard(
-            dashboard_title="test-custom-only",
-            slug="test-slug-custom",
-            editors=subjects_from_users([self.get_user("admin")]),
-        )
-        db.session.add(dashboard)
-        db.session.flush()
-
-        custom_tag = Tag(name="critical", type=TagType.custom)
-        db.session.add(custom_tag)
-        db.session.flush()
-
-        tagged_obj = TaggedObject(
-            tag_id=custom_tag.id,
-            object_id=dashboard.id,
-            object_type="dashboard",
-        )
-        db.session.add(tagged_obj)
-        db.session.commit()
-
-        try:
-            # 1. MODEL: dashboard.tags returns ALL tags
-            all_tags = dashboard.tags
-            all_tag_names = [t.name for t in all_tags]
-            assert "critical" in all_tag_names, "Should include custom tag"
-            assert any(t.name.startswith("editor:") for t in all_tags), (
-                "Should include editor tags"
-            )
-            assert any(t.name.startswith("type:") for t in all_tags), (
-                "Should include type tags"
-            )
-
-            # 2. MODEL: dashboard.custom_tags returns ONLY custom tags
-            custom_only = dashboard.custom_tags
-            custom_tag_names = [t.name for t in custom_only]
-            assert "critical" in custom_tag_names, "Should include custom tag"
-            assert not any(t.name.startswith("editor:") for t in custom_only), (
-                f"custom_tags should NOT include editor tags, got: {custom_tag_names}"
-            )
-            assert not any(t.name.startswith("type:") for t in custom_only), (
-                f"custom_tags should NOT include type tags, got: {custom_tag_names}"
-            )
-            assert len(custom_only) < len(all_tags), "Should filter out implicit tags"
-
-            # Verify all tags in custom_tags have type=custom
-            for tag in custom_only:
-                assert tag.type == TagType.custom, (
-                    f"Tag {tag.name} has type {tag.type}, expected TagType.custom"
-                )
-
-            # 3. API: With config=True, API returns ONLY custom tags
-            rv = self.client.get("api/v1/dashboard/")
-            data = json.loads(rv.data.decode("utf-8"))
-
-            assert rv.status_code == 200
-            test_dash = next(
-                (d for d in data["result"] if d["id"] == dashboard.id), None
-            )
-            assert test_dash is not None
-            # API returns "tags" (get_list override renames custom_tags→tags)
-            assert "tags" in test_dash, (
-                f"Response should have tags, got: {test_dash.keys()}"
-            )
-
-            # API should return ONLY custom tags
-            api_tag_names = [t["name"] for t in test_dash["tags"]]
-            assert "critical" in api_tag_names, "API should include custom tag"
-            assert not any(
-                t["name"].startswith("editor:") for t in test_dash["tags"]
-            ), f"API should NOT include editor tags, got: {api_tag_names}"
-            assert not any(t["name"].startswith("type:") for t in test_dash["tags"]), (
-                f"API should NOT include type tags, got: {api_tag_names}"
-            )
-            assert len(test_dash["tags"]) == 1, (
-                f"API should return only 1 custom tag, "
-                f"got {len(test_dash['tags'])}: {api_tag_names}"
-            )
-        finally:
-            db.session.delete(dashboard)
-            db.session.commit()
-            db.session.delete(custom_tag)
-            db.session.commit()

tests/integration_tests/superset_test_config.py

@@ -183,7 +183,4 @@ CUSTOM_TEMPLATE_PROCESSORS = {
 
 PRESERVE_CONTEXT_ON_EXCEPTION = False
 
-# Dashboard API: Return only custom tags (performance optimization)
-DASHBOARD_LIST_CUSTOM_TAGS_ONLY = True
-
 print("Loaded TEST config for INTEGRATION tests")

tests/integration_tests/tagging_tests.py

@@ -1,294 +0,0 @@
-# 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
-
-from superset.connectors.sqla.models import SqlaTable
-from superset.extensions import db
-from superset.models.core import FavStar
-from superset.models.dashboard import Dashboard
-from superset.models.slice import Slice
-from superset.models.sql_lab import SavedQuery
-from superset.tags.models import TaggedObject
-from superset.utils.core import DatasourceType
-from superset.utils.database import get_main_database
-from tests.integration_tests.base_tests import SupersetTestCase
-from tests.integration_tests.conftest import with_feature_flags
-from tests.integration_tests.fixtures.tags import (
-    with_tagging_system_feature,  # noqa: F401
-)
-
-
-class TestTagging(SupersetTestCase):
-    def query_tagged_object_table(self):
-        query = db.session.query(TaggedObject).all()
-        return query
-
-    def clear_tagged_object_table(self):
-        db.session.query(TaggedObject).delete()
-        db.session.commit()
-
-    @pytest.mark.usefixtures("with_tagging_system_feature")
-    def test_dataset_tagging(self):
-        """
-        Test to make sure that when a new dataset is created,
-        a corresponding tag in the tagged_objects table
-        is created
-        """
-
-        # Remove all existing rows in the tagged_object table
-        self.clear_tagged_object_table()
-
-        # Test to make sure nothing is in the tagged_object table
-        assert [] == self.query_tagged_object_table()
-
-        # Create a dataset and add it to the db
-        test_dataset = SqlaTable(
-            table_name="foo",
-            schema=None,
-            editors=[],
-            database=get_main_database(),
-            sql=None,
-            extra='{"certification": 1}',
-        )
-        db.session.add(test_dataset)
-        db.session.commit()
-
-        # Test to make sure that a dataset tag was added to the tagged_object table
-        tags = self.query_tagged_object_table()
-        assert 1 == len(tags)
-        assert "ObjectType.dataset" == str(tags[0].object_type)
-        assert test_dataset.id == tags[0].object_id
-
-        # Cleanup the db
-        db.session.delete(test_dataset)
-        db.session.commit()
-
-        # Test to make sure the tag is deleted when the associated object is deleted
-        assert [] == self.query_tagged_object_table()
-
-    @pytest.mark.usefixtures("with_tagging_system_feature")
-    def test_chart_tagging(self):
-        """
-        Test to make sure that when a new chart is created,
-        a corresponding tag in the tagged_objects table
-        is created
-        """
-
-        # Remove all existing rows in the tagged_object table
-        self.clear_tagged_object_table()
-
-        # Test to make sure nothing is in the tagged_object table
-        assert [] == self.query_tagged_object_table()
-
-        # Create a chart and add it to the db
-        test_chart = Slice(
-            slice_name="test_chart",
-            datasource_type=DatasourceType.TABLE,
-            viz_type="bubble",
-            datasource_id=1,
-        )
-        db.session.add(test_chart)
-        db.session.commit()
-
-        # Test to make sure that a chart tag was added to the tagged_object table
-        tags = self.query_tagged_object_table()
-        assert 1 == len(tags)
-        assert "ObjectType.chart" == str(tags[0].object_type)
-        assert test_chart.id == tags[0].object_id
-
-        # Cleanup the db
-        db.session.delete(test_chart)
-        db.session.commit()
-
-        # Test to make sure the tag is deleted when the associated object is deleted
-        assert [] == self.query_tagged_object_table()
-
-    @pytest.mark.usefixtures("with_tagging_system_feature")
-    def test_dashboard_tagging(self):
-        """
-        Test to make sure that when a new dashboard is created,
-        a corresponding tag in the tagged_objects table
-        is created
-        """
-
-        # Remove all existing rows in the tagged_object table
-        self.clear_tagged_object_table()
-
-        # Test to make sure nothing is in the tagged_object table
-        assert [] == self.query_tagged_object_table()
-
-        # Create a dashboard and add it to the db
-        test_dashboard = Dashboard()
-        test_dashboard.dashboard_title = "test_dashboard"
-        test_dashboard.slug = "test_slug"
-        test_dashboard.published = True
-
-        db.session.add(test_dashboard)
-        db.session.commit()
-
-        # Test to make sure that a dashboard tag was added to the tagged_object table
-        tags = self.query_tagged_object_table()
-        assert 1 == len(tags)
-        assert "ObjectType.dashboard" == str(tags[0].object_type)
-        assert test_dashboard.id == tags[0].object_id
-
-        # Cleanup the db
-        db.session.delete(test_dashboard)
-        db.session.commit()
-
-        # Test to make sure the tag is deleted when the associated object is deleted
-        assert [] == self.query_tagged_object_table()
-
-    @pytest.mark.usefixtures("with_tagging_system_feature")
-    def test_saved_query_tagging(self):
-        """
-        Test to make sure that when a new saved query is
-        created, a corresponding tag in the tagged_objects
-        table is created
-        """
-
-        # Remove all existing rows in the tagged_object table
-        self.clear_tagged_object_table()
-
-        # Test to make sure nothing is in the tagged_object table
-        assert [] == self.query_tagged_object_table()
-
-        # Create a saved query and add it to the db
-        test_saved_query = SavedQuery(label="test saved query")
-        db.session.add(test_saved_query)
-        db.session.commit()
-
-        # Test to make sure that a saved query tag was added to the tagged_object table
-        tags = self.query_tagged_object_table()
-
-        assert 2 == len(tags)
-
-        assert "ObjectType.query" == str(tags[0].object_type)
-        assert "editor:None" == str(tags[0].tag.name)
-        assert "TagType.editor" == str(tags[0].tag.type)
-        assert test_saved_query.id == tags[0].object_id
-
-        assert "ObjectType.query" == str(tags[1].object_type)
-        assert "type:query" == str(tags[1].tag.name)
-        assert "TagType.type" == str(tags[1].tag.type)
-        assert test_saved_query.id == tags[1].object_id
-
-        # Cleanup the db
-        db.session.delete(test_saved_query)
-        db.session.commit()
-
-        # Test to make sure the tag is deleted when the associated object is deleted
-        assert [] == self.query_tagged_object_table()
-
-    @pytest.mark.usefixtures("with_tagging_system_feature")
-    def test_favorite_tagging(self):
-        """
-        Test to make sure that when a new favorite object is
-        created, a corresponding tag in the tagged_objects
-        table is created
-        """
-
-        # Remove all existing rows in the tagged_object table
-        self.clear_tagged_object_table()
-
-        # Test to make sure nothing is in the tagged_object table
-        assert [] == self.query_tagged_object_table()
-
-        # Create a favorited object and add it to the db
-        test_saved_query = FavStar(user_id=1, class_name="slice", obj_id=1)
-        db.session.add(test_saved_query)
-        db.session.commit()
-
-        # Test to make sure that a favorited object tag was added to the tagged_object table  # noqa: E501
-        tags = self.query_tagged_object_table()
-        assert 1 == len(tags)
-        assert "ObjectType.chart" == str(tags[0].object_type)
-        assert test_saved_query.obj_id == tags[0].object_id
-
-        # Cleanup the db
-        db.session.delete(test_saved_query)
-        db.session.commit()
-
-        # Test to make sure the tag is deleted when the associated object is deleted
-        assert [] == self.query_tagged_object_table()
-
-    @with_feature_flags(TAGGING_SYSTEM=False)
-    def test_tagging_system(self):
-        """
-        Test to make sure that when the TAGGING_SYSTEM
-        feature flag is false, that no tags are created
-        """
-
-        # Remove all existing rows in the tagged_object table
-        self.clear_tagged_object_table()
-
-        # Test to make sure nothing is in the tagged_object table
-        assert [] == self.query_tagged_object_table()
-
-        # Create a dataset and add it to the db
-        test_dataset = SqlaTable(
-            table_name="foo",
-            schema=None,
-            editors=[],
-            database=get_main_database(),
-            sql=None,
-            extra='{"certification": 1}',
-        )
-
-        # Create a chart and add it to the db
-        test_chart = Slice(
-            slice_name="test_chart",
-            datasource_type=DatasourceType.TABLE,
-            viz_type="bubble",
-            datasource_id=1,
-        )
-
-        # Create a dashboard and add it to the db
-        test_dashboard = Dashboard()
-        test_dashboard.dashboard_title = "test_dashboard"
-        test_dashboard.slug = "test_slug"
-        test_dashboard.published = True
-
-        # Create a saved query and add it to the db
-        test_saved_query = SavedQuery(label="test saved query")
-
-        # Create a favorited object and add it to the db
-        test_favorited_object = FavStar(user_id=1, class_name="slice", obj_id=1)
-
-        db.session.add(test_dataset)
-        db.session.add(test_chart)
-        db.session.add(test_dashboard)
-        db.session.add(test_saved_query)
-        db.session.add(test_favorited_object)
-        db.session.commit()
-
-        # Test to make sure that no tags were added to the tagged_object table
-        tags = self.query_tagged_object_table()
-        assert 0 == len(tags)
-
-        # Cleanup the db
-        db.session.delete(test_dataset)
-        db.session.delete(test_chart)
-        db.session.delete(test_dashboard)
-        db.session.delete(test_saved_query)
-        db.session.delete(test_favorited_object)
-        db.session.commit()
-
-        # Test to make sure all the tags are deleted when the associated objects are deleted  # noqa: E501
-        assert [] == self.query_tagged_object_table()

tests/integration_tests/tags/dao_tests.py

@@ -22,7 +22,7 @@ from superset import db
 from superset.daos.tag import TagDAO
 from superset.models.dashboard import Dashboard
 from superset.models.slice import Slice
-from superset.tags.models import ObjectType, Tag, TaggedObject
+from superset.tags.models import ObjectType, Tag, TaggedObject, TagType
 from tests.integration_tests.base_tests import SupersetTestCase
 from tests.integration_tests.constants import ADMIN_USERNAME
 from tests.integration_tests.fixtures.tags import (
@@ -336,3 +336,47 @@ class TestTagsDAO(SupersetTestCase):
             .first()
         )
         assert tagged_object is None
+
+    @pytest.mark.usefixtures("with_tagging_system_feature")
+    def test_tagged_object_cleanup_on_dashboard_delete(self):
+        """Deleting a tagged dashboard cleans up its tagged_object rows.
+
+        Regression guard for ObjectUpdater.after_delete, the one part of the
+        old auto-tagging event-listener machinery still registered: unlike
+        editor:/type:/favorited_by: generation (removed, see TagType's
+        docstring), this cleanup applies to every tag on the object -- custom
+        tags included -- and nothing else removes these rows, since
+        TaggedObject.object_id carries no foreign key (see its column
+        comment).
+        """
+        dashboard = Dashboard(
+            dashboard_title="tag cleanup test", slug="tag-cleanup-test"
+        )
+        db.session.add(dashboard)
+        db.session.commit()
+        dashboard_id = dashboard.id
+
+        tag = self.insert_tag(name="cleanup_test_tag", tag_type=TagType.custom)
+        self.insert_tagged_object(
+            tag_id=tag.id, object_id=dashboard_id, object_type=ObjectType.dashboard
+        )
+
+        def tagged_object_count() -> int:
+            return (
+                db.session.query(TaggedObject)
+                .filter(
+                    TaggedObject.object_type == ObjectType.dashboard.name,
+                    TaggedObject.object_id == dashboard_id,
+                )
+                .count()
+            )
+
+        assert tagged_object_count() == 1
+
+        db.session.delete(dashboard)
+        db.session.commit()
+
+        assert tagged_object_count() == 0
+
+        db.session.delete(tag)
+        db.session.commit()

tests/integration_tests/tasks/async_queries_tests.py

@@ -36,9 +36,6 @@ from tests.integration_tests.fixtures.birth_names_dashboard import (
     load_birth_names_data,  # noqa: F401
 )
 from tests.integration_tests.fixtures.query_context import get_query_context
-from tests.integration_tests.fixtures.tags import (
-    with_tagging_system_feature,  # noqa: F401
-)
 from tests.integration_tests.test_app import app
 
 

tests/unit_tests/dashboards/api_test.py

@@ -52,7 +52,6 @@ def mock_dashboard() -> MagicMock:
     dash.editors = []
     dash.viewers = []
     dash.tags = []
-    dash.custom_tags = []
     dash.is_managed_externally = False
     dash.uuid = None
     return dash

tests/unit_tests/views/test_custom_tags_api_mixin.py

@@ -1,109 +0,0 @@
-# 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.
-"""Tests for both renaming mechanisms in ``CustomTagsOptimizationMixin``.
-
-The schema-level ``data_key`` rename only applies to the API's default list
-schema; requests that pass ``select_columns`` make FAB build a fresh schema
-on the fly, so for those the ``pre_get_list`` response rewrite is what keeps
-the public ``tags`` name. Both paths need coverage.
-"""
-
-from typing import Any
-
-from marshmallow import fields, Schema
-
-from superset.views.custom_tags_api_mixin import CustomTagsOptimizationMixin
-
-
-class BaseApi:
-    """Stub for the FAB ``ModelRestApi`` base. The mixin chains via
-    ``super()``, so the stub must define the hooks the mixin overrides."""
-
-    list_model_schema: Schema
-    pre_get_list_calls: int = 0
-
-    def _init_model_schemas(self) -> None:
-        self.list_model_schema = Schema.from_dict(
-            {"custom_tags": fields.List(fields.String())}
-        )()
-
-    def pre_get_list(self, _data: dict[str, Any]) -> None:
-        self.pre_get_list_calls += 1
-
-
-class CustomTagsApi(CustomTagsOptimizationMixin, BaseApi):
-    _custom_tags_only = True
-
-
-class UnoptimizedTagsApi(CustomTagsOptimizationMixin, BaseApi):
-    _custom_tags_only = False
-
-
-def test_custom_tags_schema_uses_public_tags_name() -> None:
-    api = CustomTagsApi()
-
-    api._init_model_schemas()
-
-    assert api.list_model_schema.dump({"custom_tags": ["critical"]}) == {
-        "tags": ["critical"]
-    }
-
-
-def test_custom_tags_schema_keeps_name_when_optimization_disabled() -> None:
-    api = UnoptimizedTagsApi()
-
-    api._init_model_schemas()
-
-    assert api.list_model_schema.dump({"custom_tags": ["critical"]}) == {
-        "custom_tags": ["critical"]
-    }
-
-
-def test_pre_get_list_renames_custom_tags_when_enabled() -> None:
-    api = CustomTagsApi()
-    data: dict[str, Any] = {
-        "result": [
-            {"id": 1, "custom_tags": [{"name": "critical"}]},
-            {"id": 2},
-        ]
-    }
-
-    api.pre_get_list(data)
-
-    assert data["result"][0] == {"id": 1, "tags": [{"name": "critical"}]}
-    assert data["result"][1] == {"id": 2}
-    assert api.pre_get_list_calls == 1
-
-
-def test_pre_get_list_keeps_custom_tags_when_disabled() -> None:
-    api = UnoptimizedTagsApi()
-    data: dict[str, Any] = {"result": [{"id": 1, "custom_tags": []}]}
-
-    api.pre_get_list(data)
-
-    assert data["result"][0] == {"id": 1, "custom_tags": []}
-    assert api.pre_get_list_calls == 1
-
-
-def test_pre_get_list_tolerates_missing_result_key() -> None:
-    api = CustomTagsApi()
-    data: dict[str, Any] = {"count": 0}
-
-    api.pre_get_list(data)
-
-    assert data == {"count": 0}
-    assert api.pre_get_list_calls == 1