fix(datasets): drop colliding child uuid on the overwrite path too

Addresses review findings on the metric/column uuid export fix.

- helpers.py: the collision guard only ran in the `not obj` (INSERT)
  branch, but the UPDATE branch still did `setattr(obj, "uuid", incoming)`
  unconditionally. Re-importing a clone bundle with overwrite=True matched
  the child by (table_id, name) and wrote the original's uuid onto the
  clone's child, failing with `UNIQUE constraint failed: sql_metrics.uuid`.
  Move the guard above both branches so an incoming uuid owned by another
  row is dropped on INSERT and UPDATE alike.
- helpers.py: also drop an explicit `uuid: null`. The child schemas accept
  it, and on the overwrite UPDATE it would persist a literal NULL over an
  existing child's uuid — silently, since the column is nullable and
  `unique` permits repeated NULLs — orphaning every folder leaf pointing
  at that child.
- helpers.py: document why an ambiguous `MultipleResultsFound` child match
  stays a hard error. Preferring the uuid match would rename the
  uuid-matched row while the name-matched row still holds that name,
  trading a clear failure for an opaque `(table_id, <name>)` violation at
  the next flush; callers own the recovery contract.
- import_test.py: regression tests for the overwrite re-import and the
  null-uuid paths, and stop overclaiming what the `folders` equality
  assertion proves (that JSON round-trips either way; the uuid assertions
  are the real gate).

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

UPDATING.md

@@ -24,6 +24,8 @@ assists people when migrating to a new version.
 
 ## Next
 
+- [42393](https://github.com/apache/superset/pull/42393): Exported dataset YAML now carries a `uuid` for each metric and column so that custom folder assignments (which reference metrics/columns by UUID) survive an import into another workspace. As with `folders` and `currency_code_column`, these files fail schema validation (`Unknown field: uuid`) when imported into Superset releases that predate this change; regenerate or hand-edit exports for older targets in mixed-version fleets.
+
 ### Principal listing APIs now honour related-field filters
 
 Two authorization-related listing behaviors changed for API clients. Neither

superset/datasets/schemas.py

@@ -284,6 +284,7 @@ class ImportV1ColumnSchema(Schema):
     description = fields.String(allow_none=True)
     python_date_format = fields.String(allow_none=True)
     datetime_format = fields.String(allow_none=True)
+    uuid = fields.UUID(allow_none=True)
 
 
 class ImportMetricCurrencySchema(Schema):
@@ -326,6 +327,7 @@ class ImportV1MetricSchema(Schema):
     currency = CurrencyField(ImportMetricCurrencySchema, allow_none=True)
     extra = fields.Dict(allow_none=True)
     warning_text = fields.String(allow_none=True)
+    uuid = fields.UUID(allow_none=True)
 
 
 class ImportV1DatasetSchema(Schema):

superset/models/helpers.py

@@ -507,6 +507,20 @@ class ImportExportMixin(UUIDMixin):
             obj_query = db.session.query(cls).filter(and_(*filters))
             obj = obj_query.one_or_none()
         except MultipleResultsFound:
+            # Now that children carry a ``uuid``, this match is a disjunction
+            # (name-match OR uuid-match), so a payload child can match one DB row
+            # by name and a different one by uuid — e.g. metrics renamed/swapped
+            # in the UI, then an older export re-imported with overwrite.
+            #
+            # That stays a hard error on purpose. Resolving it by preferring the
+            # uuid match would rename the uuid-matched row while the name-matched
+            # row still holds that name, trading this clear failure for an opaque
+            # ``UNIQUE constraint failed: (table_id, <name>)`` at the next flush.
+            # Callers own the recovery contract instead — see the legacy
+            # NULL-schema handling in
+            # ``superset/commands/dataset/importers/v1/utils.py`` (including its
+            # ``deleted_at`` rollback) and the ``continue`` in
+            # ``superset/commands/importers/v1/examples.py`` (issue #16051).
             logger.error(
                 "Error importing %s \n %s \n %s",
                 cls.__name__,
@@ -516,6 +530,30 @@ class ImportExportMixin(UUIDMixin):
             )
             raise
 
+        # A child ``uuid`` is globally unique, but the match above is scoped to
+        # this ``parent`` (and also matches on name). Importing a config as a
+        # clone — e.g. a dataset re-imported under an edited uuid while the
+        # original still exists — can leave the incoming child uuid owned by a
+        # different row. Writing it, whether on INSERT (new obj) or on an
+        # overwrite UPDATE (obj matched by name), would violate the ``uuid``
+        # unique constraint at flush. Drop the incoming uuid whenever it belongs
+        # to some other row so ``UUIDMixin`` keeps/assigns a distinct one,
+        # reverting to the pre-uuid-export behavior for that clone.
+        if parent is not None and "uuid" in dict_rep:
+            if dict_rep["uuid"] is None:
+                # The child import schemas accept ``uuid: null``. On the
+                # overwrite UPDATE that would write a literal NULL over an
+                # existing child's uuid — silently, since the column is nullable
+                # and ``unique`` permits repeated NULLs — leaving a child no
+                # folder can reference and no export can round-trip.
+                del dict_rep["uuid"]
+            else:
+                uuid_owner = (
+                    db.session.query(cls).filter(cls.uuid == dict_rep["uuid"]).first()
+                )
+                if uuid_owner is not None and uuid_owner is not obj:
+                    del dict_rep["uuid"]
+
         if not obj:
             is_new_obj = True
             # Create new DB object
@@ -615,6 +653,7 @@ class ImportExportMixin(UUIDMixin):
                             recursive=recursive,
                             include_parent_ref=include_parent_ref,
                             include_defaults=include_defaults,
+                            export_uuids=export_uuids,
                         )
                         for child in getattr(self, cld)
                     ],

tests/integration_tests/datasets/commands_tests.py

@@ -95,6 +95,14 @@ class TestExportDatasetsCommand(SupersetTestCase):
         type_map = {
             column.column_name: str(column.type) for column in example_dataset.columns
         }
+        # column/metric UUIDs are exported so folder references survive import;
+        # they are assigned dynamically, so build lookups by name.
+        column_uuid_map = {
+            column.column_name: str(column.uuid) for column in example_dataset.columns
+        }
+        metric_uuid_map = {
+            metric.metric_name: str(metric.uuid) for metric in example_dataset.metrics
+        }
 
         assert metadata == {
             "cache_timeout": None,
@@ -115,6 +123,7 @@ class TestExportDatasetsCommand(SupersetTestCase):
                     "advanced_data_type": None,
                     "verbose_name": None,
                     "extra": None,
+                    "uuid": column_uuid_map["source"],
                 },
                 {
                     "column_name": "target",
@@ -127,6 +136,7 @@ class TestExportDatasetsCommand(SupersetTestCase):
                     "is_dttm": False,
                     "python_date_format": None,
                     "type": type_map["target"],
+                    "uuid": column_uuid_map["target"],
                     "advanced_data_type": None,
                     "verbose_name": None,
                     "extra": None,
@@ -145,6 +155,7 @@ class TestExportDatasetsCommand(SupersetTestCase):
                     "advanced_data_type": None,
                     "verbose_name": None,
                     "extra": None,
+                    "uuid": column_uuid_map["value"],
                 },
             ],
             "database_uuid": str(example_db.uuid),
@@ -165,6 +176,7 @@ class TestExportDatasetsCommand(SupersetTestCase):
                     "metric_type": "count",
                     "verbose_name": "COUNT(*)",
                     "warning_text": None,
+                    "uuid": metric_uuid_map["count"],
                 },
                 {
                     "currency": None,
@@ -176,6 +188,7 @@ class TestExportDatasetsCommand(SupersetTestCase):
                     "metric_type": None,
                     "verbose_name": None,
                     "warning_text": None,
+                    "uuid": metric_uuid_map["sum__value"],
                 },
             ],
             "folders": None,

tests/unit_tests/datasets/commands/export_test.py

@@ -41,10 +41,27 @@ def test_export(session: Session) -> None:
     db.session.flush()
 
     columns = [
-        TableColumn(column_name="ds", is_dttm=1, type="TIMESTAMP"),
-        TableColumn(column_name="user_id", type="INTEGER"),
-        TableColumn(column_name="revenue", type="INTEGER"),
-        TableColumn(column_name="expenses", type="INTEGER"),
+        TableColumn(
+            column_name="ds",
+            is_dttm=1,
+            type="TIMESTAMP",
+            uuid=UUID("00000000-0000-0000-0000-000000000006"),
+        ),
+        TableColumn(
+            column_name="user_id",
+            type="INTEGER",
+            uuid=UUID("00000000-0000-0000-0000-000000000007"),
+        ),
+        TableColumn(
+            column_name="revenue",
+            type="INTEGER",
+            uuid=UUID("00000000-0000-0000-0000-000000000008"),
+        ),
+        TableColumn(
+            column_name="expenses",
+            type="INTEGER",
+            uuid=UUID("00000000-0000-0000-0000-000000000009"),
+        ),
         TableColumn(
             column_name="profit",
             type="INTEGER",
@@ -211,6 +228,7 @@ metrics:
   extra:
     warning_markdown: null
   warning_text: null
+  uuid: 00000000-0000-0000-0000-000000000004
 columns:
 - column_name: profit
   verbose_name: null
@@ -226,6 +244,7 @@ columns:
   datetime_format: null
   extra:
     certified_by: User
+  uuid: 00000000-0000-0000-0000-000000000005
 - column_name: ds
   verbose_name: null
   is_dttm: 1
@@ -239,6 +258,7 @@ columns:
   python_date_format: null
   datetime_format: null
   extra: null
+  uuid: 00000000-0000-0000-0000-000000000006
 - column_name: user_id
   verbose_name: null
   is_dttm: false
@@ -252,19 +272,7 @@ columns:
   python_date_format: null
   datetime_format: null
   extra: null
-- column_name: expenses
-  verbose_name: null
-  is_dttm: false
-  is_active: true
-  type: INTEGER
-  advanced_data_type: null
-  groupby: true
-  filterable: true
-  expression: null
-  description: null
-  python_date_format: null
-  datetime_format: null
-  extra: null
+  uuid: 00000000-0000-0000-0000-000000000007
 - column_name: revenue
   verbose_name: null
   is_dttm: false
@@ -278,6 +286,21 @@ columns:
   python_date_format: null
   datetime_format: null
   extra: null
+  uuid: 00000000-0000-0000-0000-000000000008
+- column_name: expenses
+  verbose_name: null
+  is_dttm: false
+  is_active: true
+  type: INTEGER
+  advanced_data_type: null
+  groupby: true
+  filterable: true
+  expression: null
+  description: null
+  python_date_format: null
+  datetime_format: null
+  extra: null
+  uuid: 00000000-0000-0000-0000-000000000009
 version: 1.0.0
 database_uuid: {database.uuid}
 """,

tests/unit_tests/datasets/commands/importers/v1/import_test.py

@@ -26,6 +26,7 @@ from unittest.mock import Mock, patch
 from urllib import request
 
 import pytest
+import yaml
 from flask import current_app
 from flask_appbuilder.security.sqla.models import Role, User
 from pytest_mock import MockerFixture
@@ -253,6 +254,288 @@ def test_import_dataset(mocker: MockerFixture, session: Session) -> None:
     assert sqla_table.database.id == database.id
 
 
+def test_export_import_round_trip_preserves_metric_folder_membership(
+    mocker: MockerFixture, session: Session
+) -> None:
+    """
+    A metric (or column) assigned to a custom folder must stay in that folder
+    after the dataset is exported and imported into another workspace.
+
+    Folder leaves reference metrics/columns by UUID. If the export drops the
+    metric/column UUIDs, the importer recreates them with fresh random UUIDs
+    while the ``folders`` JSON still points at the originals — so the metric can
+    no longer be matched to its folder and is re-homed to the default folder.
+    This exercises the full export -> import round trip and asserts the
+    imported metric/column keep the UUIDs the folder leaves reference.
+    """
+    from superset.commands.dataset.export import ExportDatasetsCommand
+    from superset.connectors.sqla.models import SqlMetric
+
+    mocker.patch.object(security_manager, "can_access", return_value=True)
+
+    engine = db.session.get_bind()
+    SqlaTable.metadata.create_all(engine)  # pylint: disable=no-member
+
+    # --- source workspace: a dataset with a metric + column pinned to a custom
+    # folder, referenced by UUID ---
+    source_db = Database(database_name="source_db", sqlalchemy_uri="sqlite://")
+    db.session.add(source_db)
+    db.session.flush()
+
+    metric_uuid = uuid.UUID("00000000-0000-0000-0000-0000000000a1")
+    column_uuid = uuid.UUID("00000000-0000-0000-0000-0000000000b2")
+    folder_uuid = uuid.UUID("00000000-0000-0000-0000-0000000000c3")
+
+    sqla_table = SqlaTable(
+        table_name="my_table",
+        database=source_db,
+        columns=[
+            TableColumn(column_name="profit", type="INTEGER", uuid=column_uuid),
+        ],
+        metrics=[
+            SqlMetric(metric_name="cnt", expression="COUNT(*)", uuid=metric_uuid),
+        ],
+        folders=[
+            {
+                "uuid": str(folder_uuid),
+                "type": "folder",
+                "name": "Custom",
+                "children": [
+                    {"uuid": str(metric_uuid), "type": "metric"},
+                    {"uuid": str(column_uuid), "type": "column"},
+                ],
+            },
+        ],
+    )
+    db.session.add(sqla_table)
+    db.session.flush()
+
+    # --- export (command-level YAML payload) ---
+    config = yaml.safe_load(ExportDatasetsCommand._file_content(sqla_table))  # pylint: disable=protected-access
+
+    # The exported metric/column must carry their UUIDs so the folder
+    # references survive the round trip.
+    assert config["metrics"][0].get("uuid") == str(metric_uuid)
+    assert any(col.get("uuid") == str(column_uuid) for col in config["columns"])
+
+    # The import schema must accept and preserve those UUIDs; without the schema
+    # fields it would reject them as unknown and the round trip would break.
+    loaded = ImportV1DatasetSchema().load(config)
+    assert loaded["metrics"][0]["uuid"] == metric_uuid
+    assert any(col["uuid"] == column_uuid for col in loaded["columns"])
+
+    # --- import into another workspace ---
+    # Model a separate workspace: drop the source dataset so its UUIDs are free
+    # (a fresh workspace has never seen them), then import against a fresh
+    # database and a brand-new dataset UUID so the importer creates the
+    # metric/column anew from the exported payload.
+    db.session.delete(sqla_table)
+    db.session.flush()
+
+    target_db = Database(database_name="target_db", sqlalchemy_uri="sqlite://")
+    db.session.add(target_db)
+    db.session.flush()
+    config["database_id"] = target_db.id
+    config["uuid"] = str(uuid.uuid4())
+
+    imported = import_dataset(config)
+
+    # The metric/column must be recreated with their original UUIDs so the
+    # typed folder leaves still resolve to them — i.e. they stay in the custom
+    # folder rather than being re-homed to the default one.
+    imported_metric = next(m for m in imported.metrics if m.metric_name == "cnt")
+    assert imported_metric.uuid == metric_uuid
+    imported_column = next(c for c in imported.columns if c.column_name == "profit")
+    assert imported_column.uuid == column_uuid
+
+    # The folders JSON is stored verbatim (it round-trips with or without the
+    # fix); the uuid assertions above are the real gate that its leaves still
+    # resolve to the imported children. This just documents the expected shape.
+    assert imported.folders == [
+        {
+            "uuid": str(folder_uuid),
+            "type": "folder",
+            "name": "Custom",
+            "children": [
+                {"uuid": str(metric_uuid), "type": "metric"},
+                {"uuid": str(column_uuid), "type": "column"},
+            ],
+        },
+    ]
+
+
+def test_import_dataset_clone_with_duplicate_child_uuid_gets_fresh_uuid(
+    mocker: MockerFixture, session: Session
+) -> None:
+    """
+    Cloning a dataset by importing its config under a new dataset UUID must not
+    fail when the original — and its metric/column UUIDs — still exist.
+
+    Metric/column UUIDs are globally unique but matched only within their parent
+    on import, so an unchanged child UUID under a *new* dataset would otherwise
+    violate the unique constraint on INSERT. The importer drops the colliding
+    UUID and lets a fresh one be assigned, so the clone imports cleanly (the
+    pre-uuid-export behavior for that workflow).
+    """
+    mocker.patch.object(security_manager, "can_access", return_value=True)
+
+    engine = db.session.get_bind()
+    SqlaTable.metadata.create_all(engine)  # pylint: disable=no-member
+
+    database = Database(database_name="my_database", sqlalchemy_uri="sqlite://")
+    db.session.add(database)
+    db.session.flush()
+
+    metric_uuid = "00000000-0000-0000-0000-0000000000d4"
+    column_uuid = "00000000-0000-0000-0000-0000000000e5"
+    config = {
+        "table_name": "my_table",
+        "uuid": str(uuid.uuid4()),
+        "metrics": [
+            {"metric_name": "cnt", "expression": "COUNT(*)", "uuid": metric_uuid},
+        ],
+        "columns": [
+            {"column_name": "profit", "type": "INTEGER", "uuid": column_uuid},
+        ],
+        "database_uuid": database.uuid,
+        "database_id": database.id,
+    }
+
+    original = import_dataset(copy.deepcopy(config))
+    assert str(original.metrics[0].uuid) == metric_uuid
+    assert str(original.columns[0].uuid) == column_uuid
+
+    # Clone: same child UUIDs, but a new dataset UUID and name (as a user editing
+    # the exported config to duplicate the dataset would produce).
+    clone_config = copy.deepcopy(config)
+    clone_config["table_name"] = "my_table_clone"
+    clone_config["uuid"] = str(uuid.uuid4())
+
+    clone = import_dataset(clone_config)
+
+    # The clone imports without hitting the unique constraint, and its children
+    # receive fresh UUIDs while the original keeps its own.
+    assert clone.id != original.id
+    assert [m.metric_name for m in clone.metrics] == ["cnt"]
+    assert [c.column_name for c in clone.columns] == ["profit"]
+    assert str(clone.metrics[0].uuid) != metric_uuid
+    assert str(clone.columns[0].uuid) != column_uuid
+    assert str(original.metrics[0].uuid) == metric_uuid
+    assert str(original.columns[0].uuid) == column_uuid
+
+
+def test_import_dataset_clone_overwrite_reimport_keeps_fresh_child_uuid(
+    mocker: MockerFixture, session: Session
+) -> None:
+    """
+    Overwriting a cloned dataset with its own bundle must not resurrect the
+    original's child UUIDs.
+
+    On the overwrite path children sync with ``sync=["columns", "metrics"]`` and
+    are matched within the parent by name, so re-importing the clone bundle
+    (which still carries the original's metric/column UUIDs) would otherwise
+    ``setattr`` those UUIDs onto the clone's children and violate the global
+    unique constraint at flush. The importer drops any incoming child UUID owned
+    by a different row on the UPDATE branch too, so the overwrite succeeds and
+    the clone's children keep the fresh UUIDs assigned on first import.
+    """
+    mocker.patch.object(security_manager, "can_access", return_value=True)
+
+    engine = db.session.get_bind()
+    SqlaTable.metadata.create_all(engine)  # pylint: disable=no-member
+
+    database = Database(database_name="my_database", sqlalchemy_uri="sqlite://")
+    db.session.add(database)
+    db.session.flush()
+
+    metric_uuid = "00000000-0000-0000-0000-0000000000d6"
+    column_uuid = "00000000-0000-0000-0000-0000000000e7"
+    config = {
+        "table_name": "my_table",
+        "uuid": str(uuid.uuid4()),
+        "metrics": [
+            {"metric_name": "cnt", "expression": "COUNT(*)", "uuid": metric_uuid},
+        ],
+        "columns": [
+            {"column_name": "profit", "type": "INTEGER", "uuid": column_uuid},
+        ],
+        "database_uuid": database.uuid,
+        "database_id": database.id,
+    }
+
+    original = import_dataset(copy.deepcopy(config))
+
+    # Clone under a new dataset UUID and name; its children get fresh UUIDs
+    # because the originals still exist.
+    clone_config = copy.deepcopy(config)
+    clone_config["table_name"] = "my_table_clone"
+    clone_config["uuid"] = str(uuid.uuid4())
+    clone = import_dataset(copy.deepcopy(clone_config))
+    fresh_metric_uuid = str(clone.metrics[0].uuid)
+    fresh_column_uuid = str(clone.columns[0].uuid)
+    assert fresh_metric_uuid != metric_uuid
+    assert fresh_column_uuid != column_uuid
+
+    # Re-import the same clone bundle with overwrite=True. The children match by
+    # (table_id, name), and the bundle still carries the original's UUIDs; the
+    # guard must keep this from writing them onto the clone's children.
+    reimported = import_dataset(copy.deepcopy(clone_config), overwrite=True)
+
+    assert reimported.id == clone.id
+    assert [m.metric_name for m in reimported.metrics] == ["cnt"]
+    assert [c.column_name for c in reimported.columns] == ["profit"]
+    # The clone keeps its fresh child UUIDs and the original is untouched.
+    assert str(reimported.metrics[0].uuid) == fresh_metric_uuid
+    assert str(reimported.columns[0].uuid) == fresh_column_uuid
+    assert str(original.metrics[0].uuid) == metric_uuid
+    assert str(original.columns[0].uuid) == column_uuid
+
+
+def test_import_dataset_null_child_uuid_keeps_existing(
+    mocker: MockerFixture, session: Session
+) -> None:
+    """
+    An explicit ``uuid: null`` must not wipe an existing child's UUID.
+
+    The child import schemas accept ``uuid=None``. Without the guard the
+    overwrite path would ``setattr`` that ``None`` onto the matched child and
+    persist a literal NULL — the column is nullable and ``unique`` permits
+    repeated NULLs, so it would fail silently and orphan every folder leaf
+    pointing at that child.
+    """
+    mocker.patch.object(security_manager, "can_access", return_value=True)
+
+    engine = db.session.get_bind()
+    SqlaTable.metadata.create_all(engine)  # pylint: disable=no-member
+
+    database = Database(database_name="my_database", sqlalchemy_uri="sqlite://")
+    db.session.add(database)
+    db.session.flush()
+
+    metric_uuid = "00000000-0000-0000-0000-0000000000f8"
+    config: dict[str, Any] = {
+        "table_name": "my_table",
+        "uuid": str(uuid.uuid4()),
+        "metrics": [
+            {"metric_name": "cnt", "expression": "COUNT(*)", "uuid": metric_uuid},
+        ],
+        "columns": [],
+        "database_uuid": database.uuid,
+        "database_id": database.id,
+    }
+    dataset = import_dataset(copy.deepcopy(config))
+    assert str(dataset.metrics[0].uuid) == metric_uuid
+
+    # Re-import the same bundle with the child uuid explicitly nulled.
+    nulled = copy.deepcopy(config)
+    nulled["metrics"][0]["uuid"] = None
+    reimported = import_dataset(nulled, overwrite=True)
+
+    assert reimported.id == dataset.id
+    assert reimported.metrics[0].uuid is not None
+    assert str(reimported.metrics[0].uuid) == metric_uuid
+
+
 def test_import_dataset_no_folder(mocker: MockerFixture, session: Session) -> None:
     """
     Test importing a dataset that was exported without folders.