mikebridge commented on code in PR #41076: URL: https://github.com/apache/superset/pull/41076#discussion_r3574959386
########## scripts/seed_junction_load.py: ########## @@ -0,0 +1,679 @@ +#!/usr/bin/env python3 +# 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. +# +# ---------------------------------------------------------------------- +# Stress-test data generator for the composite-PK migration (sc-105349). +# +# Bulk-inserts synthetic parent rows and many-to-many junction rows for +# the eight association tables that the composite-PK migration touches. +# Useful for measuring migration runtime at varying scales — run this at +# 100K / 1M / 5M / 10M rows and time the migration at each scale to +# verify the O(N log N) extrapolation. +# +# Idempotent: rerunning with the same target is a no-op; rerunning with +# a higher target adds rows up to the new total. Batched bulk INSERTs +# (10K rows per statement) make it fast on Postgres, MySQL, and SQLite. +# +# Usage (inside the Superset container): +# +# docker exec superset-superset-1 \\ +# /app/.venv/bin/python /app/scripts/seed_junction_load.py \\ +# --dashboard-slices 1000000 \\ +# --slice-user 100000 \\ +# --dashboard-user 100000 +# +# Run with no flags for the defaults shown below. Use ``--dry-run`` to +# print the planned inserts without writing anything. +# +# The script connects via Superset's standard ``DATABASE_*`` env vars +# (or ``SUPERSET__SQLALCHEMY_DATABASE_URI`` if set), so it works +# automatically inside the Superset container regardless of which +# metadata DB backend is in use. + +from __future__ import annotations + +import argparse +import logging +import os +import sys +import time +from contextlib import contextmanager +from typing import Iterator +from uuid import uuid4 + +import sqlalchemy as sa +from sqlalchemy.engine import Connection, Engine + +logger = logging.getLogger("seed_junction_load") + +# Bulk INSERT batch size. Larger values = fewer statements but more memory. +BATCH = 10_000 + +# Default per-junction-table target row counts. Tuned to mimic the shape +# of a large multi-team Superset install. Override via CLI flags. +DEFAULTS: dict[str, int] = { + "dashboard_slices": 1_000_000, + "slice_user": 100_000, + "dashboard_user": 100_000, + "dashboard_roles": 10_000, +} + +# (junction_table, fk1_col, fk2_col, parent1_table, parent2_table) +# parents reference id columns; we generate (fk1, fk2) pairs by sampling +# from the parents' existing IDs. +JUNCTIONS: list[tuple[str, str, str, str, str]] = [ + ("dashboard_slices", "dashboard_id", "slice_id", "dashboards", "slices"), + ("slice_user", "user_id", "slice_id", "ab_user", "slices"), + ("dashboard_user", "user_id", "dashboard_id", "ab_user", "dashboards"), + ("dashboard_roles", "dashboard_id", "role_id", "dashboards", "ab_role"), +] + +# Junction tables that originally carried ``UNIQUE(fk1, fk2)`` and therefore +# cannot accept duplicate ``(fk1, fk2)`` pairs even on the pre-migration +# (downgrade) schema. The other JUNCTIONS allow duplicates pre-migration. +# Only ``dashboard_slices`` is listed: the migration's other UNIQUE table +# (``report_schedule_user``) is not in JUNCTIONS — this script doesn't seed +# it — so listing it here would imply coverage that doesn't exist. Add it +# alongside a JUNCTIONS entry if that table ever gets seeded. +JUNCTIONS_WITH_UNIQUE: set[str] = {"dashboard_slices"} + + +# ---------------------------------------------------------------------- +# Connection setup +# ---------------------------------------------------------------------- + + +def build_engine() -> Engine: + """Build a SQLAlchemy engine from Superset env vars.""" + if uri := os.environ.get("SUPERSET__SQLALCHEMY_DATABASE_URI"): + logger.info("Using SUPERSET__SQLALCHEMY_DATABASE_URI from env") + return sa.create_engine(uri) + + try: + dialect = os.environ["DATABASE_DIALECT"] + user = os.environ["DATABASE_USER"] + password = os.environ["DATABASE_PASSWORD"] + host = os.environ["DATABASE_HOST"] + port = os.environ["DATABASE_PORT"] + db = os.environ["DATABASE_DB"] + except KeyError as exc: + sys.exit( + f"Missing env var {exc}; either set DATABASE_DIALECT/USER/PASSWORD/" + f"HOST/PORT/DB or SUPERSET__SQLALCHEMY_DATABASE_URI before running." + ) + + uri = f"{dialect}://{user}:{password}@{host}:{port}/{db}" + logger.info( + "Built URI from DATABASE_* env vars (dialect=%s, host=%s)", dialect, host + ) + return sa.create_engine(uri) + + +# ---------------------------------------------------------------------- +# Helpers +# ---------------------------------------------------------------------- + + +def uuid_value(dialect_name: str) -> bytes | str: + """Return a UUID in the form the active dialect expects. + + MySQL stores UUIDs as ``BINARY(16)`` (16 raw bytes); Postgres has a + native ``UUID`` type that accepts strings; SQLite stores them as + BLOB/TEXT and accepts either. Branching here keeps the seed script + backend-agnostic without depending on Superset's custom column types. + """ + if dialect_name.startswith("mysql"): + return uuid4().bytes + return str(uuid4()) + + +@contextmanager +def time_phase(name: str) -> Iterator[None]: + """Log elapsed wall time for a named phase.""" + start = time.monotonic() + logger.info("[%s] starting", name) + try: + yield + finally: + elapsed = time.monotonic() - start + logger.info("[%s] done in %.2fs", name, elapsed) + + +def count_rows(conn: Connection, table: str) -> int: + return conn.scalar(sa.text(f"SELECT COUNT(*) FROM {table}")) or 0 # noqa: S608 + + +def existing_ids(conn: Connection, table: str, limit: int | None = None) -> list[int]: + sql = f"SELECT id FROM {table} ORDER BY id" # noqa: S608 + if limit is not None: + sql += f" LIMIT {limit}" + return [row[0] for row in conn.execute(sa.text(sql))] + + +# ---------------------------------------------------------------------- +# Parent seeders +# +# Each function ensures the named parent table has at least ``target`` +# rows by inserting synthetic ones with minimal-but-valid columns. +# Returns nothing; subsequent code reads back IDs via ``existing_ids``. +# ---------------------------------------------------------------------- + + +def seed_dashboards(conn: Connection, target: int, dry_run: bool) -> None: + current = count_rows(conn, "dashboards") + if current >= target: + logger.info( + "dashboards: %d rows (target %d) — no insert needed", current, target + ) + return + needed = target - current + logger.info("dashboards: %d → %d (+%d)", current, target, needed) + if dry_run: + return + + dialect = conn.engine.dialect.name + sql = sa.text( + "INSERT INTO dashboards (uuid, dashboard_title, slug, published) " + "VALUES (:uuid, :title, :slug, :published)" + ) + for batch_start in range(0, needed, BATCH): + rows = [ + { + "uuid": uuid_value(dialect), + "title": f"seed_dashboard_{current + i}", + "slug": f"seed-dashboard-{current + i}-{uuid4().hex[:8]}", + "published": False, + } + for i in range(batch_start, min(batch_start + BATCH, needed)) + ] + conn.execute(sql, rows) + logger.info(" dashboards: inserted %d / %d", batch_start + len(rows), needed) + + +def seed_dbs(conn: Connection, dry_run: bool) -> int: + """Ensure at least one row exists in ``dbs`` (parent of ``tables``). + Returns the id to use as ``database_id`` when seeding ``tables``.""" + ids = existing_ids(conn, "dbs", limit=1) + if ids: + return ids[0] + if dry_run: + return -1 # placeholder + dialect = conn.engine.dialect.name + logger.info("dbs: inserting one synthetic database (no rows present)") + conn.execute( + sa.text( + "INSERT INTO dbs (uuid, database_name, sqlalchemy_uri, expose_in_sqllab) " + "VALUES (:uuid, :name, :uri, :expose)" + ), + { + "uuid": uuid_value(dialect), + "name": f"seed_db_{uuid4().hex[:8]}", + "uri": "sqlite:///seed.db", + "expose": False, + }, + ) + return existing_ids(conn, "dbs", limit=1)[0] + + +def seed_tables(conn: Connection, target: int, dry_run: bool) -> None: + current = count_rows(conn, "tables") + if current >= target: + logger.info("tables: %d rows (target %d) — no insert needed", current, target) + return + needed = target - current + logger.info("tables: %d → %d (+%d)", current, target, needed) + if dry_run: + return + + database_id = seed_dbs(conn, dry_run=False) + dialect = conn.engine.dialect.name + sql = sa.text( + "INSERT INTO tables (uuid, table_name, database_id) " + "VALUES (:uuid, :name, :db_id)" + ) + for batch_start in range(0, needed, BATCH): + rows = [ + { + "uuid": uuid_value(dialect), + "name": f"seed_table_{current + i}", + "db_id": database_id, + } + for i in range(batch_start, min(batch_start + BATCH, needed)) + ] + conn.execute(sql, rows) + logger.info(" tables: inserted %d / %d", batch_start + len(rows), needed) + + +def seed_slices(conn: Connection, target: int, dry_run: bool) -> None: + current = count_rows(conn, "slices") + if current >= target: + logger.info("slices: %d rows (target %d) — no insert needed", current, target) + return + needed = target - current + logger.info("slices: %d → %d (+%d)", current, target, needed) + if dry_run: + return + + # Slices reference tables.id; ensure at least one ``tables`` row exists + # so the FK is satisfiable (datasource_id is nullable but we set it for + # realism). The migration test doesn't care, but a real Superset that + # re-renders these slices does. + seed_tables(conn, target=1, dry_run=False) + table_id = existing_ids(conn, "tables", limit=1)[0] + dialect = conn.engine.dialect.name + sql = sa.text( + "INSERT INTO slices " + "(uuid, slice_name, datasource_id, datasource_type, viz_type) " + "VALUES (:uuid, :name, :ds_id, :ds_type, :viz)" + ) + for batch_start in range(0, needed, BATCH): + rows = [ + { + "uuid": uuid_value(dialect), + "name": f"seed_slice_{current + i}", + "ds_id": table_id, + "ds_type": "table", + "viz": "table", + } + for i in range(batch_start, min(batch_start + BATCH, needed)) + ] + conn.execute(sql, rows) + logger.info(" slices: inserted %d / %d", batch_start + len(rows), needed) + + +def seed_users(conn: Connection, target: int, dry_run: bool) -> None: + current = count_rows(conn, "ab_user") + if current >= target: + logger.info("ab_user: %d rows (target %d) — no insert needed", current, target) + return + needed = target - current + logger.info("ab_user: %d → %d (+%d)", current, target, needed) + if dry_run: + return + + sql = sa.text( + "INSERT INTO ab_user (first_name, last_name, username, email, active) " + "VALUES (:first, :last, :username, :email, :active)" + ) + for batch_start in range(0, needed, BATCH): + rows = [ + { + "first": "seed", + "last": f"user_{current + i}", + "username": f"seed_user_{current + i}_{uuid4().hex[:8]}", + "email": f"seed_user_{current + i}_{uuid4().hex[:8]}@example.invalid", + "active": True, + } + for i in range(batch_start, min(batch_start + BATCH, needed)) + ] + conn.execute(sql, rows) + logger.info(" ab_user: inserted %d / %d", batch_start + len(rows), needed) + + +def seed_roles(conn: Connection, target: int, dry_run: bool) -> None: + current = count_rows(conn, "ab_role") + if current >= target: + logger.info("ab_role: %d rows (target %d) — no insert needed", current, target) + return + needed = target - current + logger.info("ab_role: %d → %d (+%d)", current, target, needed) + if dry_run: + return + + sql = sa.text("INSERT INTO ab_role (name) VALUES (:name)") + for batch_start in range(0, needed, BATCH): + rows = [ + {"name": f"seed_role_{current + i}_{uuid4().hex[:8]}"} + for i in range(batch_start, min(batch_start + BATCH, needed)) + ] + conn.execute(sql, rows) + logger.info(" ab_role: inserted %d / %d", batch_start + len(rows), needed) + + +# ---------------------------------------------------------------------- +# Junction seeder +# ---------------------------------------------------------------------- + + +def _load_existing_pairs( + conn: Connection, junction: str, fk1_col: str, fk2_col: str +) -> set[tuple[int, int]]: + """Load existing ``(fk1, fk2)`` pairs from a junction table into a set. + + Used so the seeder can skip them when generating new pairs (junction + tables enforce uniqueness on the FK pair). Memory is ~32 bytes/tuple + on CPython, so 10M existing pairs is ~320MB — acceptable for a dev + machine. The junction / column names come from ``JUNCTIONS``, not + user input, so the f-string interpolation is safe. + """ + sql_text = f"SELECT {fk1_col}, {fk2_col} FROM {junction}" # noqa: S608 + return {(row[0], row[1]) for row in conn.execute(sa.text(sql_text))} + + +def _generate_new_pairs( + p1_ids: list[int], + p2_ids: list[int], + existing_pairs: set[tuple[int, int]], +) -> Iterator[tuple[int, int]]: + """Yield ``(fk1, fk2)`` pairs from the parent1 × parent2 cross-product + that are not already in ``existing_pairs``.""" + for fk1 in p1_ids: + for fk2 in p2_ids: + if (fk1, fk2) not in existing_pairs: + yield (fk1, fk2) + + +def seed_junction( + conn: Connection, + junction: str, + fk1_col: str, + fk2_col: str, + parent1: str, + parent2: str, + target: int, + dry_run: bool, +) -> None: + """Bulk-insert junction rows up to ``target`` rows total. + + Generates ``(fk1, fk2)`` pairs by walking the cross-product of + parent1 IDs × parent2 IDs in row-major order, skipping pairs that + already exist. Walking the cross-product deterministically keeps + the script replayable: re-running with the same target is a no-op, + and re-running with a higher target appends new pairs in a stable + order regardless of how many runs preceded. + """ + current = count_rows(conn, junction) + if current >= target: + logger.info( + "%s: %d rows (target %d) — no insert needed", junction, current, target + ) + return + needed = target - current + logger.info("%s: %d → %d (+%d)", junction, current, target, needed) + if dry_run: + return + + p1_ids = existing_ids(conn, parent1) + p2_ids = existing_ids(conn, parent2) + max_pairs = len(p1_ids) * len(p2_ids) + if max_pairs < target: + sys.exit( + f"Cannot reach {target} rows in {junction}: " + f"only {max_pairs} unique pairs available " + f"({len(p1_ids)} × {len(p2_ids)}). " + f"Increase parent targets and rerun." + ) + + existing_pairs: set[tuple[int, int]] = ( + _load_existing_pairs(conn, junction, fk1_col, fk2_col) if current > 0 else set() + ) + if existing_pairs: + logger.info( + " %s: loaded %d existing pairs into avoidance set", + junction, + len(existing_pairs), + ) + + insert_sql = sa.text( + f"INSERT INTO {junction} ({fk1_col}, {fk2_col}) " # noqa: S608 + f"VALUES (:fk1, :fk2)" + ) + + inserted = 0 + batch: list[dict[str, int]] = [] + for fk1, fk2 in _generate_new_pairs(p1_ids, p2_ids, existing_pairs): + batch.append({"fk1": fk1, "fk2": fk2}) + inserted += 1 + if len(batch) == BATCH or inserted == needed: + conn.execute(insert_sql, batch) + logger.info(" %s: inserted %d / %d", junction, inserted, needed) + batch = [] + if inserted == needed: + return + if inserted < needed: + sys.exit( + f"Ran out of unique pairs at {inserted}/{needed} for {junction} " + f"(parents have {len(p1_ids)} × {len(p2_ids)} = {max_pairs} pairs, " + f"{len(existing_pairs)} already present)" + ) + + +# ---------------------------------------------------------------------- +# Orchestration +# ---------------------------------------------------------------------- + + +def _compute_parent_requirements(targets: dict[str, int]) -> dict[str, int]: + """For each parent table, return the minimum row count needed so that + parent1 × parent2 ≥ target for every junction it participates in. + + Allocates ceil(sqrt(target)) rows per parent, balanced across the two + parents of each junction. The actual junction seeder will then walk + the cross-product to produce the target number of unique pairs. + """ + parent_req: dict[str, int] = {} + for junction, _, _, p1, p2 in JUNCTIONS: + target = targets.get(junction, 0) + if target == 0: + continue + sqrt_n = int(target**0.5) + 1 + parent_req[p1] = max(parent_req.get(p1, 0), sqrt_n) + parent_req[p2] = max(parent_req.get(p2, 0), sqrt_n) + return parent_req + + +def _seed_parents(conn: Connection, parent_req: dict[str, int], dry_run: bool) -> None: + """Seed parent tables in dependency order: + independent parents (ab_user, ab_role) first, then dashboards / slices / + tables (which transitively depend on dbs, seeded inside seed_tables).""" + if "ab_user" in parent_req: + seed_users(conn, parent_req["ab_user"], dry_run) + if "ab_role" in parent_req: + seed_roles(conn, parent_req["ab_role"], dry_run) + if "dashboards" in parent_req: + seed_dashboards(conn, parent_req["dashboards"], dry_run) + if "slices" in parent_req: + seed_slices(conn, parent_req["slices"], dry_run) + if "tables" in parent_req: + seed_tables(conn, parent_req["tables"], dry_run) + + +def _seed_all_junctions( + conn: Connection, targets: dict[str, int], dry_run: bool +) -> None: + for junction, fk1, fk2, p1, p2 in JUNCTIONS: + target = targets.get(junction, 0) + if target == 0: + continue + with time_phase(f"junction:{junction}"): + seed_junction(conn, junction, fk1, fk2, p1, p2, target, dry_run) + + +def inject_duplicates( + conn: Connection, + junction: str, + fk1_col: str, + fk2_col: str, + pct: float, + dry_run: bool, +) -> None: + """Insert duplicate ``(fk1, fk2)`` rows on a non-UNIQUE junction table. + + Used to stress-test the migration's ``_dedupe_by_min_id`` phase, which + is otherwise a no-op on cleanly-seeded data. Computes ``count = + current_rows * pct / 100`` and inserts that many rows by re-sampling + existing ``(fk1, fk2)`` pairs in row-major order. The synthetic + duplicates land on top of distinct existing pairs (one duplicate per + distinct pair, then wraps), so the migration's dedupe finds and + deletes them. + + **Pre-condition: the table must NOT have UNIQUE on (fk1, fk2)**, i.e., + the schema must be the pre-migration shape (after running + ``superset db downgrade``). On the post-migration schema the composite + PK rejects duplicates and this function will error. + """ + if pct == 0: + return + current = count_rows(conn, junction) + count = int(current * pct / 100) + if count == 0: + logger.info( + "%s: 0 duplicates to inject (current=%d, pct=%g)", + junction, + current, + pct, + ) + return + logger.info( + "%s: injecting %d duplicate rows (%g%% of %d existing)", + junction, + count, + pct, + current, + ) + if dry_run: + return + + select_sql = sa.text( + f"SELECT {fk1_col}, {fk2_col} FROM {junction} ORDER BY id LIMIT :n" # noqa: S608 + ) + sample = conn.execute(select_sql, {"n": count}).fetchall() + if not sample: + logger.warning("%s: no rows to duplicate (table is empty)", junction) + return + + insert_sql = sa.text( + f"INSERT INTO {junction} ({fk1_col}, {fk2_col}) " # noqa: S608 + f"VALUES (:fk1, :fk2)" + ) + inserted = 0 + while inserted < count: + batch: list[dict[str, int]] = [] + while len(batch) < BATCH and inserted < count: + row = sample[inserted % len(sample)] + batch.append({"fk1": row[0], "fk2": row[1]}) + inserted += 1 + conn.execute(insert_sql, batch) + logger.info(" %s: injected %d / %d duplicates", junction, inserted, count) + + +def _inject_dirty_data(conn: Connection, dirty_pct: float, dry_run: bool) -> None: + """Inject duplicate rows on every non-UNIQUE seeded junction. + + The two tables that originally carried ``UNIQUE(fk1, fk2)`` are + skipped because their composite-PK successor (and their pre-migration + UNIQUE constraint) both reject duplicate inserts. + """ + if dirty_pct == 0: + return + for junction, fk1, fk2, _, _ in JUNCTIONS: + if junction in JUNCTIONS_WITH_UNIQUE: + logger.info( + "%s: skipping duplicate injection (table has UNIQUE on FK pair)", + junction, + ) + continue + with time_phase(f"dirty:{junction}"): + inject_duplicates(conn, junction, fk1, fk2, dirty_pct, dry_run) + + +def run(targets: dict[str, int], dry_run: bool, dirty_duplicates_pct: float) -> None: + engine = build_engine() + with engine.begin() as conn: + parent_req = _compute_parent_requirements(targets) + logger.info("Required parent row counts: %s", parent_req) + + with time_phase("parents"): + _seed_parents(conn, parent_req, dry_run) + + with time_phase("junctions"): + _seed_all_junctions(conn, targets, dry_run) + + if dirty_duplicates_pct > 0: + with time_phase("dirty-duplicates"): + _inject_dirty_data(conn, dirty_duplicates_pct, dry_run) + + +# ---------------------------------------------------------------------- +# CLI +# ---------------------------------------------------------------------- + + +def main() -> None: + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + for table, default in DEFAULTS.items(): + parser.add_argument( + f"--{table.replace('_', '-')}", + type=int, + default=default, + help=f"target row count for {table} (default: {default:,})", + ) + parser.add_argument( + "--dry-run", + "-n", + action="store_true", + help="print planned inserts without writing to the DB", + ) + parser.add_argument( + "--dirty-duplicates-pct", + type=float, + default=0, + help=( + "after seeding distinct pairs, inject this percentage of duplicate " + "rows on each non-UNIQUE junction (slice_user, dashboard_user, " + "dashboard_roles). Stress-tests the migration's _dedupe_by_min_id " + "phase. Requires 2bee73611e32 to NOT be applied: un-apply it by " + "downgrading to its parent (`superset db downgrade <down_revision>`, " + "where <down_revision> is read from the 2bee73611e32 migration " + "file) — the post-migration composite PK rejects duplicates and " + "this will error. Default: 0 (no duplicates)." + ), + ) + parser.add_argument( + "--verbose", + "-v", + action="store_true", + help="increase log verbosity", + ) + args = parser.parse_args() + + logging.basicConfig( + level=logging.DEBUG if args.verbose else logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%H:%M:%S", + ) + + targets = {table: getattr(args, table) for table in DEFAULTS} Review Comment: Addressed by subsequent commits; the flagged code is no longer current after the rebase onto master. Resolving as outdated. ########## superset/charts/api.py: ########## @@ -437,9 +496,29 @@ def put(self, pk: int) -> Response: # This validates custom Schema with custom validations except ValidationError as error: return self.response_400(message=error.messages) + + # Live version identifiers before the update (empty + query-free when + # ``ENABLE_VERSIONING_CAPTURE`` is off, so this stays inert under the + # kill-switch). + old_info = current_entity_version_info(Slice, pk) Review Comment: Addressed by subsequent commits; the flagged code is no longer current after the rebase onto master. Resolving as outdated. ########## superset/charts/api.py: ########## @@ -437,9 +496,29 @@ def put(self, pk: int) -> Response: # This validates custom Schema with custom validations except ValidationError as error: return self.response_400(message=error.messages) + + # Live version identifiers before the update (empty + query-free when + # ``ENABLE_VERSIONING_CAPTURE`` is off, so this stays inert under the + # kill-switch). + old_info = current_entity_version_info(Slice, pk) + try: changed_model = UpdateChartCommand(pk, item).run() - response = self.response(200, id=changed_model.id, result=item) + new_info = current_entity_version_info( + Slice, changed_model.id, changed_model.uuid + ) + response = self.response( Review Comment: Addressed by subsequent commits; the flagged code is no longer current after the rebase onto master. Resolving as outdated. ########## superset/commands/dashboard/update.py: ########## @@ -59,23 +59,31 @@ def __init__(self, model_id: int, data: dict[str, Any]): def run(self) -> Model: self.validate() assert self._model is not None - self.process_tab_diff() - self.process_native_filter_diff() - - # Update tags - if (tags := self._properties.pop("tags", None)) is not None: - update_tags(ObjectType.dashboard, self._model.id, self._model.tags, tags) - - # Re-serialize position_json to escape 4-byte Unicode characters - if position_json := self._properties.get("position_json"): - self._properties["position_json"] = json.dumps(json.loads(position_json)) - - dashboard = DashboardDAO.update(self._model, self._properties) - if self._properties.get("json_metadata"): - DashboardDAO.set_dash_metadata( - dashboard, - data=json.loads(self._properties.get("json_metadata", "{}")), - ) + # Suppress autoflush during the update body so that Continuum's + # before_flush baseline listener does not fire mid-operation while + # the session is only partially populated. + with db.session.no_autoflush: + self.process_tab_diff() + self.process_native_filter_diff() + + # Update tags + if (tags := self._properties.pop("tags", None)) is not None: + update_tags( + ObjectType.dashboard, self._model.id, self._model.tags, tags + ) + + # Re-serialize position_json to escape 4-byte Unicode characters + if position_json := self._properties.get("position_json"): + self._properties["position_json"] = json.dumps( + json.loads(position_json) + ) + + dashboard = DashboardDAO.update(self._model, self._properties) Review Comment: Addressed by subsequent commits; the flagged code is no longer current after the rebase onto master. Resolving as outdated. ########## superset/dashboards/api.py: ########## @@ -828,17 +887,32 @@ def put(self, pk: int) -> Response: # This validates custom Schema with custom validations except ValidationError as error: return self.response_400(message=error.messages) + + # Live version identifiers before the update (empty + query-free when + # ``ENABLE_VERSIONING_CAPTURE`` is off). + old_info = current_entity_version_info(Dashboard, pk) Review Comment: Addressed by subsequent commits; the flagged code is no longer current after the rebase onto master. Resolving as outdated. ########## superset/dashboards/api.py: ########## @@ -828,17 +887,32 @@ def put(self, pk: int) -> Response: # This validates custom Schema with custom validations except ValidationError as error: return self.response_400(message=error.messages) + + # Live version identifiers before the update (empty + query-free when + # ``ENABLE_VERSIONING_CAPTURE`` is off). + old_info = current_entity_version_info(Dashboard, pk) + try: changed_model = UpdateDashboardCommand(pk, item).run() last_modified_time = changed_model.changed_on.replace( microsecond=0 ).timestamp() + new_info = current_entity_version_info( + Dashboard, changed_model.id, changed_model.uuid + ) Review Comment: Addressed by subsequent commits; the flagged code is no longer current after the rebase onto master. Resolving as outdated. ########## superset/migrations/versions/2026-05-01_23-36_2bee73611e32_composite_pk_association_tables.py: ########## @@ -0,0 +1,580 @@ +# 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. +"""composite_pk_association_tables + +Replace the unused synthetic ``id INTEGER PRIMARY KEY`` on eight many-to-many +association tables with a composite primary key on the two FK columns. Drops +the now-redundant ``UniqueConstraint(fk1, fk2)`` on the two tables that +already carry one. Pre-flight: deletes rows with NULL FK values (six tables +allow them today) and any duplicate ``(fk1, fk2)`` rows. + +Motivated by SQLAlchemy-Continuum issue #129 (M2M restore against junction +tables with surrogate PKs); also closes the data-integrity hole where six +of the eight tables lacked DB-level uniqueness. + +Revision ID: 2bee73611e32 +Revises: 78a40c08b4be +Create Date: 2026-05-01 23:36:34.050058 + +""" + +import logging +from typing import NamedTuple + +import sqlalchemy as sa +from alembic import op +from alembic.operations.base import BatchOperations +from sqlalchemy import inspect +from sqlalchemy.engine import Connection + +# revision identifiers, used by Alembic. +revision = "2bee73611e32" +down_revision = "78a40c08b4be" Review Comment: Addressed by subsequent commits; the flagged code is no longer current after the rebase onto master. Resolving as outdated. ########## superset/migrations/versions/2026-05-01_23-36_2bee73611e32_composite_pk_association_tables.py: ########## @@ -0,0 +1,580 @@ +# 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. +"""composite_pk_association_tables + +Replace the unused synthetic ``id INTEGER PRIMARY KEY`` on eight many-to-many +association tables with a composite primary key on the two FK columns. Drops +the now-redundant ``UniqueConstraint(fk1, fk2)`` on the two tables that +already carry one. Pre-flight: deletes rows with NULL FK values (six tables +allow them today) and any duplicate ``(fk1, fk2)`` rows. + +Motivated by SQLAlchemy-Continuum issue #129 (M2M restore against junction +tables with surrogate PKs); also closes the data-integrity hole where six +of the eight tables lacked DB-level uniqueness. + +Revision ID: 2bee73611e32 +Revises: 78a40c08b4be +Create Date: 2026-05-01 23:36:34.050058 + +""" + +import logging +from typing import NamedTuple + +import sqlalchemy as sa +from alembic import op +from alembic.operations.base import BatchOperations +from sqlalchemy import inspect +from sqlalchemy.engine import Connection + +# revision identifiers, used by Alembic. +revision = "2bee73611e32" +down_revision = "78a40c08b4be" + +logger = logging.getLogger("alembic.env") + + +class AssociationTable(NamedTuple): + """A junction table being converted from surrogate-id PK to composite-FK PK.""" + + name: str + fk1: str + fk2: str + + +# Order is alphabetical by table name; deterministic for review and bisection. +AFFECTED_TABLES: list[AssociationTable] = [ + AssociationTable("dashboard_roles", "dashboard_id", "role_id"), + AssociationTable("dashboard_slices", "dashboard_id", "slice_id"), + AssociationTable("dashboard_user", "user_id", "dashboard_id"), + AssociationTable("report_schedule_user", "user_id", "report_schedule_id"), + AssociationTable("rls_filter_roles", "role_id", "rls_filter_id"), + AssociationTable("rls_filter_tables", "table_id", "rls_filter_id"), + AssociationTable("slice_user", "user_id", "slice_id"), + AssociationTable("sqlatable_user", "user_id", "table_id"), +] + +# These two tables already declare ``UniqueConstraint(fk1, fk2)`` in the model; +# the composite PK subsumes it, so the migration drops the redundant constraint. +TABLES_WITH_PRE_EXISTING_UNIQUE: set[str] = { + "dashboard_slices", + "report_schedule_user", +} + +# Documentation set: tables whose FK columns are nullable in their original +# create_table migrations (``dashboard_roles.dashboard_id`` from revision +# e11ccdd12658 is the most recent addition). ``report_schedule_user`` is the +# only affected table created with both FK columns ``NOT NULL`` and is +# intentionally absent here. This set is no longer consulted at runtime — the +# upgrade now runs the NULL-FK cleanup on every affected table because the +# DELETE is a cheap no-op when the columns are already NOT NULL, and that +# eliminates the risk of bugs from this set going stale (the +# ``dashboard_roles`` omission caught in PR review was exactly that bug). +TABLES_WITH_NULLABLE_FKS: set[str] = { + "dashboard_roles", + "dashboard_slices", + "dashboard_user", + "rls_filter_roles", + "rls_filter_tables", + "slice_user", + "sqlatable_user", +} + + +def _check_no_external_fks_to_id(conn: Connection) -> None: + """Raise ``RuntimeError`` if any foreign key in the database references one + of the eight junction-table ``id`` columns. Uses SQLAlchemy's ``Inspector`` + for dialect-agnostic introspection across PostgreSQL, MySQL, and SQLite. + + Scope limitation: ``Inspector.get_table_names()`` returns tables in the + connection's default schema only. On PostgreSQL deployments where Superset + metadata lives in a non-default schema, or on multi-schema deployments + that allow cross-schema FKs, an external FK in another schema would not + be detected. This is acceptable for the standard single-schema + deployment that Superset documents; operators with multi-schema + metadata should run the equivalent inventory query against + ``information_schema.referential_constraints`` themselves before + applying. + """ + affected = {t.name for t in AFFECTED_TABLES} + insp = inspect(conn) + for table_name in insp.get_table_names(): + if table_name in affected: + continue + for fk in insp.get_foreign_keys(table_name): + if fk["referred_table"] in affected and "id" in fk["referred_columns"]: + raise RuntimeError( + f"Cannot drop synthetic id from {fk['referred_table']}: " + f"external FK {fk.get('name', '<unnamed>')} on {table_name} " + f"references {fk['referred_table']}({fk['referred_columns']}). " + "Drop or migrate the referencing FK before applying this " + "migration." + ) + + +def _table_clause(t: AssociationTable) -> sa.sql.expression.TableClause: + """Build a lightweight SQLAlchemy ``TableClause`` for ``t`` exposing the + columns the helper queries reference (``id``, ``fk1``, ``fk2``). Used so + that the dedupe / cleanup / assert SQL can be expressed via SQLAlchemy + core constructs rather than via string interpolation.""" + return sa.table(t.name, sa.column("id"), sa.column(t.fk1), sa.column(t.fk2)) + + +def _delete_null_fk_rows(conn: Connection, t: AssociationTable) -> int: + """Delete rows where ``t.fk1`` or ``t.fk2`` is NULL on ``t.name``. + + Returns the deletion count. Required because primary-key columns must be + NOT NULL; the PK-add downstream would fail with a cryptic constraint + violation if any NULL-FK rows survived. Run unconditionally on every + affected table — see ``TABLES_WITH_NULLABLE_FKS`` above for the rationale. + """ + tbl = _table_clause(t) + stmt = sa.delete(tbl).where(sa.or_(tbl.c[t.fk1].is_(None), tbl.c[t.fk2].is_(None))) + result = conn.execute(stmt) + n = result.rowcount or 0 + if n: + logger.warning( + "Deleted %d row(s) with NULL FK from %s before composite-PK promotion", + n, + t.name, + ) + return n + + +def _dedupe_by_min_id(conn: Connection, t: AssociationTable) -> int: + """Delete duplicate ``(t.fk1, t.fk2)`` rows from ``t.name`` keeping ``MIN(id)``. + + Returns the deletion count. The ``NOT IN`` argument is wrapped in an + extra ``SELECT keep_id FROM (...) AS s`` derived table because MySQL + rejects ``DELETE FROM t WHERE id NOT IN (SELECT MIN(id) FROM t GROUP BY + ...)`` with ERROR 1093 unless the inner SELECT is materialized through + a derived table. SQLAlchemy's ``.subquery()`` produces that wrap. + + Logs a sample (up to 10) of the discarded ``(fk1, fk2, id)`` tuples at + WARN before deletion, so operators can audit which rows are dropped — + the "keep ``MIN(id)``" policy preserves the original row, which is + correct in practice but discards any later, semantically-identical + re-grants. + """ + tbl = _table_clause(t) + + keep_min = ( + sa.select(sa.func.min(tbl.c.id).label("keep_id")) + .group_by(tbl.c[t.fk1], tbl.c[t.fk2]) + .subquery("keep_min") + ) + keep_ids = sa.select(keep_min.c.keep_id) + discarded = tbl.c.id.notin_(keep_ids) + + sample_stmt = ( + sa.select(tbl.c[t.fk1], tbl.c[t.fk2], tbl.c.id).where(discarded).limit(10) + ) + sample = list(conn.execute(sample_stmt)) + + delete_stmt = sa.delete(tbl).where(discarded) + result = conn.execute(delete_stmt) + n = result.rowcount or 0 + if n: + logger.warning( + "Deduped %d duplicate row(s) from %s; sample of discarded " + "(%s, %s, id) tuples (up to 10): %s", + n, + t.name, + t.fk1, + t.fk2, + sample, + ) + return n + + +def _assert_no_duplicates(conn: Connection, t: AssociationTable) -> None: + """Raise ``RuntimeError`` if any ``(t.fk1, t.fk2)`` duplicate group remains. + + Called after ``_dedupe_by_min_id`` to surface silent dialect-dependent + dedupe failures (e.g., a MySQL syntax issue) as an actionable error + before the PK-add fires with a less-helpful constraint-violation message. + """ + tbl = _table_clause(t) + duplicate_groups = ( + sa.select(sa.literal(1)) + .select_from(tbl) + .group_by(tbl.c[t.fk1], tbl.c[t.fk2]) + .having(sa.func.count() > 1) + .subquery("duplicate_groups") + ) + count_stmt = sa.select(sa.func.count()).select_from(duplicate_groups) + if remaining := conn.scalar(count_stmt) or 0: + raise RuntimeError( + f"Dedupe failed for {t.name}: {remaining} duplicate " + f"({t.fk1}, {t.fk2}) groups remain after _dedupe_by_min_id. " + f"Check the dedupe SQL for dialect {conn.dialect.name}." + ) + + +def _build_pre_upgrade_table( + insp: sa.engine.reflection.Inspector, + t: AssociationTable, + fks: list[dict] | None = None, +) -> sa.Table: + """Build a ``Table`` object representing the pre-upgrade schema of ``t``, + explicitly *without* any redundant ``UniqueConstraint(t.fk1, t.fk2)``. + Used as ``copy_from`` to ``batch_alter_table`` so the rebuilt table + omits the unnamed UNIQUE constraint deterministically across dialects + (SQLite reflects unnamed UNIQUEs with ``name=None``, defeating the + standard ``batch_op.drop_constraint(name)`` path). + + Reflects column types and FK targets (with original FK constraint names + preserved) from the live database; only the redundant UNIQUE is omitted. + + *fks* lets a caller pass a pre-captured ``get_foreign_keys`` result. + The MySQL upgrade path drops the live FK constraints before building + this table, so re-reflecting here would only see them via the + Inspector's per-instance ``info_cache`` — an implementation detail, + not a contract. Passing the pre-drop list makes the dependency + explicit instead of relying on reflection caching. + """ + md = sa.MetaData() + if fks is None: + fks = insp.get_foreign_keys(t.name) + fks_for_col: dict[str, list[dict]] = {} + for fk in fks: + for col_name in fk["constrained_columns"]: + fks_for_col.setdefault(col_name, []).append(fk) + + cols: list[sa.Column] = [] + for c in insp.get_columns(t.name): + col_kwargs = {"nullable": c.get("nullable", True)} + if c["name"] == "id": + col_kwargs["primary_key"] = True + col_kwargs["autoincrement"] = True + fk_args = [] + for fk in fks_for_col.get(c["name"], []): + idx = fk["constrained_columns"].index(c["name"]) + target = f"{fk['referred_table']}.{fk['referred_columns'][idx]}" + options = {} + if fk.get("options", {}).get("ondelete"): + options["ondelete"] = fk["options"]["ondelete"] + if fk.get("name"): + options["name"] = fk["name"] + fk_args.append(sa.ForeignKey(target, **options)) + cols.append(sa.Column(c["name"], c["type"], *fk_args, **col_kwargs)) + return sa.Table(t.name, md, *cols) + + +def _drop_redundant_unique_by_name( + conn: Connection, insp: sa.engine.reflection.Inspector, t: AssociationTable +) -> None: + """Drop the redundant ``UNIQUE(fk1, fk2)`` constraint by its reflected + name on PostgreSQL / MySQL. + + The two tables in ``TABLES_WITH_PRE_EXISTING_UNIQUE`` carry a UNIQUE + constraint that the composite primary key subsumes. PostgreSQL and + MySQL both auto-name UNIQUE constraints (``<table>_<cols>_key`` on + Postgres, ``<table>_<col>_<n>`` or the explicit ``uq_*`` we may have + given it on MySQL), so they're reflectable by name. SQLite is + handled separately via ``recreate="always"`` + ``copy_from`` because + it reflects unnamed UNIQUEs with ``name=None``. + + No-op if no matching UNIQUE is found (defensive — re-runs after a + partial application should not error). + """ + for uc in insp.get_unique_constraints(t.name): + if set(uc.get("column_names", [])) == {t.fk1, t.fk2} and uc.get("name"): + op.drop_constraint(uc["name"], t.name, type_="unique") + return + + +# MySQL ON DELETE actions that the downgrade re-create loop is allowed +# to interpolate into raw SQL. The reflected value comes from MySQL's +# information_schema (so not user input), but a whitelist eliminates +# the "what if an unexpected value appears" question entirely. The +# four entries are the SQL-standard set; SET DEFAULT is intentionally +# excluded because InnoDB silently downgrades it to NO ACTION. +_VALID_ONDELETE_ACTIONS: frozenset[str] = frozenset( + {"CASCADE", "SET NULL", "RESTRICT", "NO ACTION"} +) + + +def _enforce_not_null_for_sqlite( + batch_op: BatchOperations, t: AssociationTable, conn: Connection +) -> None: + """Force ``NOT NULL`` on the FK columns post-PK-promotion on SQLite only. + + SQLite has a long-standing quirk: composite ``PRIMARY KEY`` does not + promote constituent columns to ``NOT NULL`` (only ``INTEGER PRIMARY KEY`` + does). PostgreSQL and MySQL implicitly promote the PK columns to + ``NOT NULL`` when the constraint is added, making the explicit + ``alter_column`` redundant there. + + Skipping the ``alter_column`` on MySQL is also functionally required: + MySQL 8 rejects ``ALTER COLUMN`` on a column that participates in a + foreign key constraint with ``ERROR 1832 (HY000): Cannot change column + 'X': used in a foreign key constraint 'Y'`` whenever the table has + data — even when the only change is ``NULL`` → ``NOT NULL`` and the + column is already part of a freshly-added composite primary key (which + InnoDB has just made implicitly ``NOT NULL`` anyway). The error fires + on populated tables but not on empty ones, which is why CI's + ``test-mysql`` shard (fresh schema) didn't catch this and a real + production-shaped install does. + + Only SQLite still needs the explicit step, and SQLite has no FK + enforcement objection. + """ + if conn.dialect.name == "sqlite": + batch_op.alter_column(t.fk1, existing_type=sa.Integer, nullable=False) + batch_op.alter_column(t.fk2, existing_type=sa.Integer, nullable=False) + + +def upgrade() -> None: + conn = op.get_bind() + _check_no_external_fks_to_id(conn) + insp = inspect(conn) + + for t in AFFECTED_TABLES: + # Resumability guard: on MySQL every DDL statement auto-commits, so + # a failure at table N of 8 leaves tables 1..N-1 already converted + # while ``alembic_version`` is still un-stamped. Without this guard + # a re-run would fail at table 1 (``drop_column("id")`` on a table + # that no longer has ``id``), and ``downgrade`` can't run either + # (the revision was never stamped) — recovery would need manual + # surgery. A converted table is identified by the absent ``id`` + # column; skipping it makes re-running the upgrade safe on every + # dialect (Postgres/SQLite wrap the migration in a transaction, so + # the guard is simply never hit there). + if "id" not in {c["name"] for c in insp.get_columns(t.name)}: + logger.info( + "%s: already converted (no surrogate id column); skipping", + t.name, + ) + continue + + # Run NULL-FK cleanup unconditionally: it is a no-op DELETE on tables + # whose FK columns are already NOT NULL (cheap), and skipping it on a + # table whose FK was nullable would leave the PK-add to fail with a + # cryptic constraint violation. Cf. ``TABLES_WITH_NULLABLE_FKS`` above + # for documentation of which tables are known to have nullable FKs. + _delete_null_fk_rows(conn, t) + _dedupe_by_min_id(conn, t) + _assert_no_duplicates(conn, t) + + # Two tables (``dashboard_slices``, ``report_schedule_user``) + # carry a redundant ``UNIQUE(fk1, fk2)`` that the composite PK + # subsumes. Three dialect-specific paths: + # + # * **PostgreSQL** — the UNIQUE constraint has a stable + # reflected name (Postgres default convention), so we + # ``DROP CONSTRAINT`` by name and then run the structural + # change as direct ALTER. This avoids the full-table copy + # that ``recreate="always"`` would trigger + # (``CREATE TABLE AS SELECT → DROP → RENAME``), holding + # ``ACCESS EXCLUSIVE`` only for the (much shorter) PK + # index build instead of the full copy duration. + # + # * **MySQL** — InnoDB binds the FK constraints to the + # redundant UNIQUE's underlying index for back-reference, + # so a direct ``DROP CONSTRAINT`` of the UNIQUE raises + # ``ERROR 1553``. Use ``recreate="always"`` to rebuild the + # table without the UNIQUE; drop the FKs first to dodge + # the ``ERROR 1826`` (duplicate FK constraint name) that + # the temp-table phase would otherwise provoke. The FKs + # are re-created automatically as part of ``copy_from``. + # + # * **SQLite** — unnamed UNIQUE constraints reflect with + # ``name=None`` and can't be dropped by name. Use + # ``recreate="always"`` + ``copy_from`` (omits UNIQUE). + # SQLite always rebuilds for PK changes anyway, so the + # recreate isn't extra cost there. + if t.name in TABLES_WITH_PRE_EXISTING_UNIQUE: + if conn.dialect.name == "postgresql": + _drop_redundant_unique_by_name(conn, insp, t) + with op.batch_alter_table(t.name) as batch_op: + batch_op.drop_column("id") + batch_op.create_primary_key(f"pk_{t.name}", [t.fk1, t.fk2]) + _enforce_not_null_for_sqlite(batch_op, t, conn) + else: + # Capture the FK list BEFORE dropping: the copy_from table + # below must embed these constraints, and re-reflecting + # after the drop only works via the Inspector's + # per-instance info_cache (see _build_pre_upgrade_table). + pre_drop_fks = insp.get_foreign_keys(t.name) + if conn.dialect.name == "mysql": + for fk in pre_drop_fks: + if fk_name := fk.get("name"): + op.drop_constraint(fk_name, t.name, type_="foreignkey") Review Comment: Addressed by subsequent commits; the flagged code is no longer current after the rebase onto master. Resolving as outdated. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
