You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@superset.apache.org by mi...@apache.org on 2026/09/16 13:35:53 UTC
(superset) 01/02: perf(deletion-retention): compute purge-audit repeat predicate once per entity set
This is an automated email from the ASF dual-hosted git repository.
aminghadersohi pushed a commit to branch sc-120950-purge-audit-window
in repository https://gitbox.apache.org/repos/asf/superset.git
commit 2c29a1d46e117fae4300b63135ae1455f7dd0052
Author: Amin Ghadersohi <am...@gmail.com>
AuthorDate: Wed Sep 16 13:10:26 2026 +0000
perf(deletion-retention): compute purge-audit repeat predicate once per entity set
Cherry-picked from apache/superset#44262 (commit 0cf3835ac0f3164b4236b1d22e2fab381ff2552b)
to stack this follow-up on its unmerged prerequisite.
---
.../commands/deletion_retention/prune_audit.py | 279 +++++++++++++++------
.../deletion_retention/prune_audit_tests.py | 176 ++++++++++++-
.../deletion_retention/test_prune_audit.py | 229 +++++++++++++++++
3 files changed, 605 insertions(+), 79 deletions(-)
diff --git a/superset/commands/deletion_retention/prune_audit.py b/superset/commands/deletion_retention/prune_audit.py
index 826d57f3743..0ffe84c14b5 100644
--- a/superset/commands/deletion_retention/prune_audit.py
+++ b/superset/commands/deletion_retention/prune_audit.py
@@ -88,11 +88,10 @@ pending or recovered row committed before the lock is honoured (a row it turned
into a survivor fails the re-check and is not deleted). The re-check is a SELECT
rather than a correlated ``WHERE`` on the DELETE because MySQL rejects a DELETE
whose subquery reads the target table (ERROR 1093); the DELETE names only the
-literal surviving ids. Because the re-check predicates are correlated
-(per-entity index probes) and scoped to at most the configured batch of ids,
-the locked work is bounded by the batch size times the history depth of the
-entities in it rather than by the table's size — a workload-dependent cost,
-not a time bound, which the batch-size setting trades against drain speed.
+literal surviving ids. The repeat check computes timestamp groups for the
+batch's entities in uncorrelated window tables; the other guards use
+correlated index probes. Lock-hold depends on the batch's entity histories,
+not a fixed time bound, which the batch-size setting trades against drain speed.
Audit creation/recovery and every pruning batch's DELETE take the same
singleton database write lock. The lock is held through commit, so an audit row
@@ -105,7 +104,7 @@ operators explicitly choose their retention policy.
from __future__ import annotations
import logging
-from collections.abc import Callable
+from collections.abc import Callable, Sequence
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from functools import partial
@@ -158,11 +157,12 @@ _STREAK_BREAKING_STATUSES: frozenset[str] = frozenset(
BATCH_SIZE: int = 50
#: Upper bound on the configurable batch size — a conservative cross-dialect
#: ceiling, not a derived limit. The locked re-check and the literal-id DELETE
-#: bind every id of the batch in one ``IN`` list alongside the predicates' own
-#: bindings, and the smallest parameter budget among supported backends
-#: (SQLite builds commonly allow 999) must hold both; 500 leaves ample room
-#: for the fixed predicate bindings and is the largest value this code has
-#: been exercised with.
+#: bind every id of the batch in one ``IN`` list. The re-check also binds at
+#: most 500 entity UUIDs and 500 entity types: ~1,500 distinct values plus
+#: scalars. Positional dialects repeat the scope lists in three derived tables,
+#: requiring ~3,500 placeholders plus scalars. SQLite development/test builds
+#: need the larger variable budget (32,766 by default from SQLite 3.32.0),
+#: rather than the legacy 999 default; custom build/runtime limits can differ.
MAX_BATCH_SIZE: int = 500
BATCH_SIZE_KEY: str = "PURGE_AUDIT_PRUNING_BATCH_SIZE"
#: Sentinel distinguishing an absent config key from an explicit ``None``.
@@ -348,7 +348,9 @@ def _in_current_streak(target: sa.FromClause, now: datetime) -> sa.ColumnElement
def _repeats_an_earlier_block(
- table: sa.FromClause, now: datetime
+ table: sa.FromClause,
+ now: datetime,
+ scope_entities: Sequence[tuple[str, str | None]] | None = None,
) -> sa.ColumnElement[bool]:
"""Whether a same-reason current-streak block precedes this row with no
change of reason in between.
@@ -372,53 +374,105 @@ def _repeats_an_earlier_block(
survivor while its streak is current. It is also exempt from operational
age-out (see :func:`_operational_candidates`), so an operator force-purge
block is retained permanently — never pruned by either category.
+
+ P is the immediately preceding distinct blocked timestamp. A row repeats
+ only when P is in the current streak and both timestamp groups contain
+ solely its reason (including all-NULL groups). LAG over timestamp groups
+ supplies P and its reason counts without a group self-join. The sc-120493
+ PostgreSQL round-1 plan materialized a groups CTE and joined on entity
+ alone before filtering ranks, comparing 36 million row pairs.
+
+ Keep the repeat-id query uncorrelated: sc-120493 measurements in
+ lock-hold-evidence/REPORT.md, Variant 2, showed MySQL repeatedly executing
+ per-row predecessor scalars. During re-check, scope blocked rows, timestamp
+ groups and boundaries with literal entity-type and UUID lists from the
+ fresh locked lookup. Their cross-product may include extra entity histories,
+ but candidacy remains restricted to the discovered ids and checked in SQL.
"""
- earlier: sa.FromClause = table.alias("earlier_block")
- between: sa.FromClause = table.alias("reason_change")
- reason_changed_between: sa.ColumnElement[bool] = sa.exists(
- sa.select(sa.literal(1))
- .select_from(between)
- .where(
- sa.and_(
- between.c.status == STATUS_BLOCKED,
- between.c.entity_type == table.c.entity_type,
- between.c.entity_uuid == table.c.entity_uuid,
- # Inclusive bounds: a differing-reason block sharing an
- # exact timestamp with either endpoint still breaks the run,
- # so a reason-transition row tied with a neighbour is
- # preserved as a run head rather than pruned as a repeat
- # (the same preserving-side tie rule the pending and evidence
- # guards use). Inclusive bounds only ever add boundaries —
- # i.e. only ever preserve more, never delete more.
- between.c.created_on >= earlier.c.created_on,
- between.c.created_on <= table.c.created_on,
- between.c.reason.is_distinct_from(table.c.reason),
- )
+ source: sa.Table = PurgeAuditLog.__table__
+ scope: list[sa.ColumnElement[bool]] = []
+ if scope_entities is not None:
+ types: list[str] = sorted({t for t, _ in scope_entities})
+ uuids: list[str] = sorted({u for _, u in scope_entities if u is not None})
+ scope = [source.c.entity_type.in_(types), source.c.entity_uuid.in_(uuids)]
+ blocked_filters: list[sa.ColumnElement[bool]] = [
+ source.c.status == STATUS_BLOCKED,
+ source.c.entity_uuid.is_not(None),
+ source.c.created_on <= now,
+ ]
+ blocked: sa.Subquery = (
+ sa.select(
+ source.c.id,
+ source.c.entity_type,
+ source.c.entity_uuid,
+ source.c.created_on,
+ source.c.reason,
+ source.c.trigger,
)
- .correlate(table, earlier)
+ .select_from(source)
+ .where(*blocked_filters, *scope)
+ .correlate(None)
+ .subquery("blocked_rows")
)
- repeats: sa.ColumnElement[bool] = sa.exists(
- sa.select(sa.literal(1))
- .select_from(earlier)
+ groups: sa.Subquery = (
+ sa.select(
+ source.c.entity_type,
+ source.c.entity_uuid,
+ source.c.created_on.label("ts"),
+ sa.func.count().label("n"),
+ sa.func.count(source.c.reason).label("n_coded"),
+ sa.func.min(source.c.reason).label("min_reason"),
+ sa.func.max(source.c.reason).label("max_reason"),
+ )
+ .select_from(source)
+ .where(*blocked_filters, *scope)
+ .group_by(source.c.entity_type, source.c.entity_uuid, source.c.created_on)
+ .correlate(None)
+ .subquery("blocked_timestamp_groups")
+ )
+ grp: sa.Subquery = (
+ sa.select(
+ groups,
+ *[
+ sa.func.lag(groups.c[name])
+ .over(
+ partition_by=(groups.c.entity_type, groups.c.entity_uuid),
+ order_by=groups.c.ts,
+ )
+ .label(f"prev_{name}")
+ for name in ("ts", "n", "n_coded", "min_reason", "max_reason")
+ ],
+ )
+ .correlate(None)
+ .subquery("preceding_groups")
+ )
+ boundary: sa.Subquery = (
+ sa.select(
+ source.c.entity_type,
+ source.c.entity_uuid,
+ sa.func.max(source.c.created_on).label("boundary_ts"),
+ )
+ .select_from(source)
.where(
- sa.and_(
- earlier.c.status == STATUS_BLOCKED,
- earlier.c.entity_type == table.c.entity_type,
- earlier.c.entity_uuid == table.c.entity_uuid,
- # ``earlier`` must itself be in the current streak; its
- # boundary scalar correlates to ``earlier`` (not ``table``).
- _in_current_streak(earlier, now),
- earlier.c.created_on < table.c.created_on,
- earlier.c.reason.is_not_distinct_from(table.c.reason),
- sa.not_(reason_changed_between),
- )
+ source.c.status.in_(_STREAK_BREAKING_STATUSES),
+ source.c.entity_uuid.is_not(None),
+ source.c.created_on <= now,
+ *scope,
)
- # Correlate the ``earlier`` EXISTS to ``table`` only. The boundary is
- # no longer a passed subquery — it is a scalar correlated inside
- # ``_in_current_streak(earlier, now)`` above, so it must not be
- # co-correlated here.
- .correlate(table)
+ .group_by(source.c.entity_type, source.c.entity_uuid)
+ .correlate(None)
+ .subquery("repeat_boundary")
)
+
+ def uniform(
+ n: sa.ColumnElement[Any],
+ n_coded: sa.ColumnElement[Any],
+ lo: sa.ColumnElement[Any],
+ hi: sa.ColumnElement[Any],
+ ) -> sa.ColumnElement[bool]:
+ """Require all NULL reasons or one shared non-NULL reason."""
+ return sa.or_(n_coded == 0, sa.and_(n_coded == n, lo == hi))
+
# A ``force`` attempt is an operator action the writer never suppresses:
# audit.py's ``_suppress_redundant_block`` only collapses consecutive
# scheduled same-reason blocks, so the pruner does not collapse a force row
@@ -430,7 +484,50 @@ def _repeats_an_earlier_block(
# force-purge blocks. A force row may still be the *earlier* anchor a later
# scheduled repeat collapses into — only the force row itself is protected
# from duplicate removal.
- return sa.and_(table.c.trigger != TRIGGER_FORCE, repeats)
+ not_force: sa.ColumnElement[bool] = blocked.c.trigger != TRIGGER_FORCE
+ repeat_ids: sa.sql.Select = (
+ sa.select(blocked.c.id)
+ .select_from(
+ blocked.join(
+ grp,
+ sa.and_(
+ grp.c.entity_type == blocked.c.entity_type,
+ grp.c.entity_uuid == blocked.c.entity_uuid,
+ grp.c.ts == blocked.c.created_on,
+ ),
+ ).outerjoin(
+ boundary,
+ sa.and_(
+ boundary.c.entity_type == blocked.c.entity_type,
+ boundary.c.entity_uuid == blocked.c.entity_uuid,
+ ),
+ )
+ )
+ .where(
+ not_force,
+ grp.c.prev_ts.is_not(None),
+ sa.or_(
+ boundary.c.boundary_ts.is_(None), grp.c.prev_ts > boundary.c.boundary_ts
+ ),
+ # Inclusive bounds: a differing-reason block sharing an
+ # exact timestamp with either endpoint still breaks the run,
+ # so a reason-transition row tied with a neighbour is
+ # preserved as a run head rather than pruned as a repeat
+ # (the same preserving-side tie rule the pending and evidence
+ # guards use). Inclusive bounds only ever add boundaries —
+ # i.e. only ever preserve more, never delete more.
+ uniform(grp.c.n, grp.c.n_coded, grp.c.min_reason, grp.c.max_reason),
+ uniform(
+ grp.c.prev_n,
+ grp.c.prev_n_coded,
+ grp.c.prev_min_reason,
+ grp.c.prev_max_reason,
+ ),
+ grp.c.prev_min_reason.is_not_distinct_from(blocked.c.reason),
+ )
+ .correlate(None)
+ )
+ return table.c.id.in_(repeat_ids)
def _preceded_by_unresolved_attempt(table: sa.Table) -> sa.ColumnElement[bool]:
@@ -493,24 +590,25 @@ def _duplicate_candidates(now: datetime, limit: int) -> sa.sql.Select:
def _duplicate_predicates(
- table: sa.FromClause, now: datetime
+ table: sa.FromClause,
+ now: datetime,
+ scope_entities: Sequence[tuple[str, str | None]] | None = None,
) -> list[sa.ColumnElement[bool]]:
"""The candidacy predicates for the blocked-duplicate category.
One list, used by BOTH the unlocked discovery select
(:func:`_duplicate_candidates`) and the coordination-locked re-check
delete (:func:`_delete_batch`), so the two can never drift — the locked
- delete re-verifies exactly what discovery selected. All predicates are
- correlated on *table* (the streak boundary is a per-row scalar, the
- repeat/pending checks are correlated ``EXISTS``), so applied to a bounded
- id set they resolve as per-entity index probes rather than a full scan.
+ delete re-verifies exactly what discovery selected. The repeat check is an
+ uncorrelated IN over histories scoped by literal entity-type and UUID
+ lists; boundary and pending checks remain correlated index probes.
"""
return [
table.c.status == STATUS_BLOCKED,
table.c.entity_uuid.is_not(None),
table.c.created_on <= now,
_in_current_streak(table, now),
- _repeats_an_earlier_block(table, now),
+ _repeats_an_earlier_block(table, now, scope_entities=scope_entities),
sa.not_(_preceded_by_unresolved_attempt(table)),
]
@@ -558,18 +656,22 @@ def _operational_candidates(
def _operational_predicates(
- table: sa.FromClause, now: datetime, cutoff: datetime
+ table: sa.FromClause,
+ now: datetime,
+ cutoff: datetime,
+ scope_entities: Sequence[tuple[str, str | None]] | None = None,
) -> list[sa.ColumnElement[bool]]:
"""The candidacy predicates for the operational-expiry category.
One list, shared by the unlocked discovery select
(:func:`_operational_candidates`) and the coordination-locked re-check
- delete, so the two cannot drift. Correlated on *table*, so a bounded id
- set resolves per-entity via index.
+ delete, so the two cannot drift. Repeats use an uncorrelated IN scoped by
+ the batch's literal entity-type and UUID lists; other guards retain
+ correlated index probes.
"""
is_survivor: sa.ColumnElement[bool] = sa.and_(
_in_current_streak(table, now),
- sa.not_(_repeats_an_earlier_block(table, now)),
+ sa.not_(_repeats_an_earlier_block(table, now, scope_entities=scope_entities)),
)
unstable_block: sa.ColumnElement[bool] = sa.and_(
table.c.status == STATUS_BLOCKED,
@@ -623,7 +725,10 @@ def _evidence_candidates(now: datetime, cutoff: datetime, limit: int) -> sa.sql.
def _evidence_predicates(
- table: sa.FromClause, now: datetime, cutoff: datetime
+ table: sa.FromClause,
+ now: datetime,
+ cutoff: datetime,
+ scope_entities: Sequence[tuple[str, str | None]] | None = None,
) -> list[sa.ColumnElement[bool]]:
"""The candidacy predicates for the evidence-expiry category.
@@ -661,7 +766,7 @@ def _evidence_predicates(
def _delete_batch(
select_candidates: Callable[[int], sa.sql.Select],
- recheck_predicates: Callable[[sa.FromClause], list[sa.ColumnElement[bool]]],
+ recheck_predicates: Callable[..., list[sa.ColumnElement[bool]]],
batch_size: int = BATCH_SIZE,
) -> tuple[int, int]:
"""Discover a candidate batch UNLOCKED, then delete it under the lock.
@@ -671,8 +776,8 @@ def _delete_batch(
caller keys "drained" on *discovered*, not *removed*, so an overlapping run
that thinned the hints cannot be misread as a fully-drained backlog.
- Discovery (the expensive age-unbounded scan with its boundary / repeat /
- pending correlated subqueries) runs WITHOUT the coordination lock and only
+ Discovery (the age-unbounded scan with repeat-group window tables and
+ correlated boundary/pending guards) runs WITHOUT the coordination lock and only
yields ≤``BATCH_SIZE`` candidate ids — a hint. The lock is then taken for a
re-check SELECT that re-applies the SAME candidacy predicates scoped to
those ids, followed by a literal-id DELETE of the survivors. The re-check
@@ -680,10 +785,11 @@ def _delete_batch(
a DELETE whose subquery reads the target table (ERROR 1093), and the
candidacy predicates read ``purge_audit_log``; the DELETE then names only
the literal surviving-id list. Both statements are scoped to the batch's
- ids, so the locked work is per-PK + per-entity index probes whose cost
- grows with the batch size and the candidate entities' history depth, not
- with the table's size (sc-116701 lock-hold-time). When discovery finds
- nothing, no lock is taken at all.
+ ids, with repeat groups scoped by literal entity-type and UUID lists read
+ under the lock before the re-check. Its cost
+ depends on batch size, history depth and the optimizer's access plan;
+ entity scoping does not guarantee index seeks or a lock-hold time bound
+ (sc-120493). When discovery finds nothing, no lock is taken at all.
The backdated-write invariant is preserved: a row that a concurrent
``write_ahead`` turned into a survivor between the unlocked discovery and
@@ -729,6 +835,20 @@ def _delete_batch(
# after the lock is held.
db.session.rollback() # pylint: disable=consider-using-transaction
acquire_coordination_lock(db.session)
+ # Scoping only: entity_type/entity_uuid are immutable facts about the hint
+ # ids, not a candidacy decision. Read them under the lock on the fresh
+ # snapshot; the SQL re-check below re-verifies every predicate (sc-118200).
+ # Literal values expose column statistics to the planner; a scope subquery
+ # or JOIN caused severe cardinality underestimates and nested loops
+ # (sc-120493). DISTINCT pairs are bounded by len(ids) <= MAX_BATCH_SIZE.
+ scope_entities: list[tuple[str, str | None]] = [
+ (row[0], row[1])
+ for row in db.session.execute(
+ sa.select(table.c.entity_type, table.c.entity_uuid)
+ .where(table.c.id.in_(ids))
+ .distinct()
+ )
+ ]
# Locked re-check as a SELECT, then a literal-id DELETE. With the lock held
# no writer can commit, and the re-check's fresh post-lock snapshot sees
# current committed state; it re-applies the SAME candidacy predicates
@@ -738,13 +858,15 @@ def _delete_batch(
# correlated WHERE on the DELETE — because MySQL rejects a DELETE whose
# subquery reads the target table (ERROR 1093), and the candidacy predicates
# read ``purge_audit_log``; the DELETE then names only the literal surviving
- # ids. Both statements are bounded to the batch's ids (per-PK + per-entity
- # index probes); the lock-hold grows with the batch size and the candidate
- # entities' history depth, which the batch-size setting trades off.
+ # ids. Both statements are bounded to the batch's ids; the lock-hold also
+ # depends on the scoped histories and the optimizer's access plan.
valid_ids: list[Any] = [
row[0]
for row in db.session.execute(
- sa.select(table.c.id).where(table.c.id.in_(ids), *recheck_predicates(table))
+ sa.select(table.c.id).where(
+ table.c.id.in_(ids),
+ *recheck_predicates(table, scope_entities=scope_entities),
+ )
)
]
if not valid_ids:
@@ -772,7 +894,8 @@ class _Category(NamedTuple):
"""One delete category: how to discover candidates and how to re-check them.
``select_candidates`` is the unlocked discovery select (LIMIT-bounded);
- ``recheck_predicates`` is the SAME candidacy as correlated ``WHERE`` clauses
+ ``recheck_predicates`` accepts the fresh entity pairs as scope_entities,
+ binds their types and UUIDs as literal lists, and supplies the SAME candidacy
for the coordination-locked, id-scoped re-check delete. Both are built from
one shared predicate list per category, so discovery and the locked gate
cannot drift.
@@ -780,7 +903,7 @@ class _Category(NamedTuple):
name: _CategoryName
select_candidates: Callable[[int], sa.sql.Select]
- recheck_predicates: Callable[[sa.FromClause], list[sa.ColumnElement[bool]]]
+ recheck_predicates: Callable[..., list[sa.ColumnElement[bool]]]
class _DrainResult(NamedTuple):
diff --git a/tests/integration_tests/deletion_retention/prune_audit_tests.py b/tests/integration_tests/deletion_retention/prune_audit_tests.py
index c7cd4e39398..a11a715d95a 100644
--- a/tests/integration_tests/deletion_retention/prune_audit_tests.py
+++ b/tests/integration_tests/deletion_retention/prune_audit_tests.py
@@ -25,7 +25,8 @@ uuid, which a ``LIKE`` on the uuid column would never match.
from __future__ import annotations
-from collections.abc import Callable
+import random
+from collections.abc import Callable, Sequence
from datetime import datetime, timedelta
from functools import partial
from typing import Any
@@ -905,3 +906,176 @@ class TestPruneAudit(SupersetTestCase):
"(permanent immortality per the force exemption)"
)
assert boundary in survivors
+
+
+def _legacy_repeats_an_earlier_block(
+ table: sa.FromClause,
+ now: datetime,
+ scope_entities: Sequence[tuple[str, str | None]] | None = None,
+) -> sa.ColumnElement[bool]:
+ """The pre-rewrite predicate, verbatim in its executable form."""
+ earlier: sa.FromClause = table.alias("earlier_block")
+ between: sa.FromClause = table.alias("reason_change")
+ reason_changed_between: sa.ColumnElement[bool] = sa.exists(
+ sa.select(sa.literal(1))
+ .select_from(between)
+ .where(
+ sa.and_(
+ between.c.status == STATUS_BLOCKED,
+ between.c.entity_type == table.c.entity_type,
+ between.c.entity_uuid == table.c.entity_uuid,
+ between.c.created_on >= earlier.c.created_on,
+ between.c.created_on <= table.c.created_on,
+ between.c.reason.is_distinct_from(table.c.reason),
+ )
+ )
+ .correlate(table, earlier)
+ )
+ repeats: sa.ColumnElement[bool] = sa.exists(
+ sa.select(sa.literal(1))
+ .select_from(earlier)
+ .where(
+ sa.and_(
+ earlier.c.status == STATUS_BLOCKED,
+ earlier.c.entity_type == table.c.entity_type,
+ earlier.c.entity_uuid == table.c.entity_uuid,
+ prune_audit._in_current_streak(earlier, now),
+ earlier.c.created_on < table.c.created_on,
+ earlier.c.reason.is_not_distinct_from(table.c.reason),
+ sa.not_(reason_changed_between),
+ )
+ )
+ .correlate(table)
+ )
+ return sa.and_(table.c.trigger != audit.TRIGGER_FORCE, repeats)
+
+
+_EQUIV_ENTITY_TYPE: str = f"{_PREFIX}equiv"
+_EQUIV_SEEDS: range = range(40) # deterministic; recorded so a failure is reproducible
+_EQUIV_STATUSES: list[str] = [STATUS_BLOCKED] * 12 + [
+ STATUS_PENDING,
+ STATUS_CONFIRMED,
+ STATUS_TARGET_ABSENT,
+ STATUS_FAILED,
+]
+_EQUIV_REASONS: list[str | None] = [_REASON_A, _REASON_B, "user_attribute", None]
+
+
+def _random_history(rng: random.Random, now: datetime) -> list[dict[str, Any]]:
+ """One randomised multi-entity audit history exercising every rule.
+
+ Per entity: 20-60 rows; timestamps step 0-3 days (a 0 step is a TIE, so
+ tied same-reason and tied differing-reason neighbours occur); reasons
+ repeat the previous one with high probability (runs) and flip otherwise,
+ including to/from ``None`` (legacy reason-less runs); statuses are mostly
+ ``blocked`` with occasional ``pending`` (unresolved-attempt boundaries),
+ ``confirmed``/``target_absent`` (streak breakers / evidence bounds) and
+ ``failed``; a few rows are ``force``-triggered. Ages span ~1 year so the
+ operational and evidence cutoffs both bite.
+ """
+ rows: list[dict[str, Any]] = []
+ for entity in range(rng.randint(2, 4)):
+ uuid_: str = f"{_PREFIX}equiv-{entity}"
+ ts: datetime = now - timedelta(days=rng.randint(200, 400))
+ reason: str | None = rng.choice(_EQUIV_REASONS)
+ for _ in range(rng.randint(20, 60)):
+ ts = ts + timedelta(days=rng.choice([0, 0, 1, 1, 2, 3]))
+ if ts >= now:
+ break
+ status: str = rng.choice(_EQUIV_STATUSES)
+ if rng.random() < 0.25:
+ reason = rng.choice(_EQUIV_REASONS)
+ rows.append(
+ {
+ "id": uuid4(),
+ "status": status,
+ "trigger": audit.TRIGGER_FORCE
+ if rng.random() < 0.08
+ else audit.TRIGGER_RETENTION,
+ "actor": audit.ACTOR_SYSTEM,
+ "entity_type": _EQUIV_ENTITY_TYPE,
+ "entity_uuid": uuid_,
+ "reason": reason if status == STATUS_BLOCKED else None,
+ "removed_dashboard_slices": 0,
+ "created_on": ts,
+ }
+ )
+ return rows
+
+
+class TestRepeatPredicateEquivalence(SupersetTestCase):
+ """The rewritten repeat predicate selects exactly what the legacy one did."""
+
+ def setUp(self) -> None:
+ super().setUp()
+ self._cleanup()
+
+ def tearDown(self) -> None:
+ self._cleanup()
+ super().tearDown()
+
+ def _cleanup(self) -> None:
+ db.session.rollback()
+ db.session.execute(
+ sa.delete(PurgeAuditLog.__table__).where(
+ PurgeAuditLog.__table__.c.entity_type == _EQUIV_ENTITY_TYPE
+ )
+ )
+ db.session.commit()
+
+ @staticmethod
+ def _candidate_sets(now: datetime) -> dict[str, set[UUID]]:
+ table: sa.Table = PurgeAuditLog.__table__
+ cutoff_op: datetime = now - timedelta(days=90)
+ cutoff_ev: datetime = now - timedelta(days=180)
+ out: dict[str, set[UUID]] = {}
+ for name, preds in (
+ ("duplicate", prune_audit._duplicate_predicates(table, now)),
+ ("operational", prune_audit._operational_predicates(table, now, cutoff_op)),
+ ("evidence", prune_audit._evidence_predicates(table, now, cutoff_ev)),
+ ):
+ out[name] = {
+ r[0]
+ for r in db.session.execute(
+ sa.select(table.c.id).where(
+ table.c.entity_type == _EQUIV_ENTITY_TYPE, *preds
+ )
+ ).all()
+ }
+ return out
+
+ def test_rewritten_repeat_predicate_matches_legacy_on_random_histories(
+ self,
+ ) -> None:
+ """Compare all candidate categories for every recorded random seed.
+
+ The rewritten predicate must equal the legacy
+ nested-EXISTS predicate — the executable form of the equivalence
+ argument, covering ties, reason flips (incl. to/from NULL), pending
+ boundaries, streak breakers, evidence bounds and force rows."""
+ now: datetime = audit.utc_now()
+ checked_rows: int = 0
+ for seed in _EQUIV_SEEDS:
+ rng: random.Random = random.Random(seed) # noqa: S311 — deterministic test data, not crypto
+ rows: list[dict[str, Any]] = _random_history(rng, now)
+ self._cleanup()
+ db.session.execute(sa.insert(PurgeAuditLog.__table__), rows)
+ db.session.commit()
+ checked_rows += len(rows)
+
+ rewritten: dict[str, set[UUID]] = self._candidate_sets(now)
+ with patch.object(
+ prune_audit,
+ "_repeats_an_earlier_block",
+ _legacy_repeats_an_earlier_block,
+ ):
+ legacy: dict[str, set[UUID]] = self._candidate_sets(now)
+
+ for category in ("duplicate", "operational", "evidence"):
+ assert rewritten[category] == legacy[category], (
+ f"seed={seed} category={category}: "
+ f"only-rewritten={len(rewritten[category] - legacy[category])} "
+ f"only-legacy={len(legacy[category] - rewritten[category])}"
+ )
+ # Guard against a generator regression that quietly checks nothing.
+ assert checked_rows > 1000
diff --git a/tests/unit_tests/commands/deletion_retention/test_prune_audit.py b/tests/unit_tests/commands/deletion_retention/test_prune_audit.py
index 08a8700d1e4..7dea9293bdc 100644
--- a/tests/unit_tests/commands/deletion_retention/test_prune_audit.py
+++ b/tests/unit_tests/commands/deletion_retention/test_prune_audit.py
@@ -29,6 +29,7 @@ from decimal import Decimal
from functools import partial
from typing import Any, Iterator
from unittest.mock import MagicMock, patch
+from uuid import UUID, uuid4
import pytest
import sqlalchemy as sa
@@ -429,3 +430,231 @@ def test_pruning_shares_the_audit_writers_clock() -> None:
from superset.commands.deletion_retention import audit
assert prune_audit.utc_now is audit.utc_now
+
+
+@pytest.mark.parametrize(
+ "dialect", [postgresql.dialect(), mysql.dialect(), sqlite.dialect()]
+)
+def test_repeat_history_scope_is_only_present_for_locked_rechecks(dialect: Any) -> None:
+ """Limit history to candidate entities only when a batch scope is supplied."""
+ table: sa.Table = prune_audit.PurgeAuditLog.__table__
+ now: datetime = datetime(2026, 1, 1)
+ scope_entities: list[tuple[str, str | None]] = [
+ ("dashboard", "b"),
+ ("chart", "a"),
+ ("chart", "a"),
+ ]
+ scoped: sa.sql.Select = sa.select(table.c.id).where(
+ *prune_audit._duplicate_predicates(table, now, scope_entities=scope_entities)
+ )
+ unscoped: sa.sql.Select = sa.select(table.c.id).where(
+ *prune_audit._duplicate_predicates(table, now)
+ )
+ scoped_sql: str = str(
+ scoped.compile(dialect=dialect, compile_kwargs={"literal_binds": True})
+ )
+ unscoped_sql: str = str(unscoped.compile(dialect=dialect))
+ assert scoped_sql.count("entity_type IN ('chart', 'dashboard')") == 3
+ assert scoped_sql.count("entity_uuid IN ('a', 'b')") == 3
+ assert "repeat_scope" not in scoped_sql
+ assert "entity_type IN" not in unscoped_sql
+ assert "entity_uuid IN" not in unscoped_sql
+
+
+def test_maximum_literal_scope_executes_with_sqlite_bind_budget() -> None:
+ """Account for repeated positional scope binds at the 500-id ceiling."""
+ engine: sa.Engine = sa.create_engine("sqlite://")
+ metadata: sa.MetaData = sa.MetaData()
+ table: sa.Table = prune_audit.PurgeAuditLog.__table__.to_metadata(metadata)
+ ids: list[UUID] = [uuid4() for _ in range(prune_audit.MAX_BATCH_SIZE)]
+ pairs: list[tuple[str, str | None]] = [
+ (f"type-{i}", f"entity-{i}") for i in range(len(ids))
+ ]
+ query: sa.sql.Select = sa.select(table.c.id).where(
+ table.c.id.in_(ids),
+ *prune_audit._duplicate_predicates(
+ table, datetime(2026, 1, 1), scope_entities=pairs
+ ),
+ )
+ compiled: sa.sql.compiler.Compiled = query.compile(
+ dialect=sqlite.dialect(), compile_kwargs={"render_postcompile": True}
+ )
+ assert 1500 < len(compiled.params) < 1550
+ assert compiled.positiontup is not None
+ assert 3500 < len(compiled.positiontup) < 3550
+ try:
+ metadata.create_all(engine)
+ with engine.connect() as connection:
+ assert list(connection.scalars(query)) == []
+ finally:
+ engine.dispose()
+
+
+def test_repeat_query_uses_lag_without_ctes() -> None:
+ """Keep the measured derived-table shape without materialized group self-joins."""
+ table: sa.Table = prune_audit.PurgeAuditLog.__table__
+ query: sa.sql.Select = sa.select(table.c.id).where(
+ *prune_audit._duplicate_predicates(
+ table, datetime(2026, 1, 1), scope_entities=[("chart", "entity")]
+ )
+ )
+ sql: str = str(query.compile(dialect=postgresql.dialect())).lower()
+ assert "with " not in sql
+ assert "lag(" in sql
+ assert "dense_rank(" not in sql
+
+
+@pytest.mark.parametrize("count", [1, 2, prune_audit.MAX_BATCH_SIZE])
+def test_delete_batch_forwards_bounded_entity_scope_to_the_locked_recheck(
+ count: int,
+) -> None:
+ """Use at most one distinct entity pair per discovered id."""
+ ids: list[UUID] = [uuid4() for _ in range(count)]
+ scope_entities: list[tuple[str, str | None]] = [
+ ("chart", str(i)) for i in range(count)
+ ]
+ select_candidates: MagicMock = MagicMock()
+ recheck_predicates: MagicMock = MagicMock(return_value=[])
+ mock_db: MagicMock
+ with (
+ patch.object(prune_audit, "db") as mock_db,
+ patch.object(prune_audit, "acquire_coordination_lock"),
+ ):
+ mock_db.session.execute.side_effect = [
+ [(value,) for value in ids],
+ scope_entities,
+ [],
+ ]
+ assert prune_audit._delete_batch(
+ select_candidates, recheck_predicates, count
+ ) == (
+ count,
+ 0,
+ )
+ recheck_predicates.assert_called_once_with(
+ prune_audit.PurgeAuditLog.__table__, scope_entities=scope_entities
+ )
+ assert len(recheck_predicates.call_args.kwargs["scope_entities"]) <= len(ids)
+ mock_db.session.rollback.assert_called_once()
+
+
+def test_delete_batch_reads_entity_scope_only_after_the_fresh_snapshot_lock() -> None:
+ """Pin discovery, fresh-snapshot scope lookup, SQL re-check and deletion order."""
+ table: sa.Table = prune_audit.PurgeAuditLog.__table__
+ ids: list[UUID] = [uuid4(), uuid4()]
+ pairs: list[tuple[str, str | None]] = [("chart", "entity"), ("chart", None)]
+ discovery: sa.sql.Select = sa.select(table.c.id).limit(2)
+ select_candidates: MagicMock = MagicMock(return_value=discovery)
+ recheck_predicates: MagicMock = MagicMock(return_value=[])
+ events: MagicMock = MagicMock()
+ mock_db: MagicMock
+ lock: MagicMock
+ with (
+ patch.object(prune_audit, "db") as mock_db,
+ patch.object(prune_audit, "acquire_coordination_lock") as lock,
+ ):
+ events.attach_mock(mock_db.session.execute, "execute")
+ events.attach_mock(mock_db.session.rollback, "rollback")
+ events.attach_mock(lock, "lock")
+ events.attach_mock(mock_db.session.commit, "commit")
+ mock_db.session.execute.side_effect = [
+ [(id_,) for id_ in ids],
+ pairs,
+ [(ids[0],)],
+ MagicMock(rowcount=1),
+ ]
+ assert prune_audit._delete_batch(select_candidates, recheck_predicates, 2) == (
+ 2,
+ 1,
+ )
+ assert [event[0] for event in events.mock_calls] == [
+ "execute",
+ "rollback",
+ "lock",
+ "execute",
+ "execute",
+ "execute",
+ "commit",
+ ]
+ lock.assert_called_once_with(mock_db.session)
+ select_candidates.assert_called_once_with(2)
+ recheck_predicates.assert_called_once_with(table, scope_entities=pairs)
+ assert mock_db.session.execute.call_args_list[0].args[0] is discovery
+ scope_select: sa.sql.Select = mock_db.session.execute.call_args_list[1].args[0]
+ scope_sql: str = str(
+ scope_select.compile(
+ dialect=postgresql.dialect(),
+ compile_kwargs={"literal_binds": True},
+ )
+ )
+ assert scope_sql.startswith(
+ "SELECT DISTINCT purge_audit_log.entity_type, purge_audit_log.entity_uuid"
+ )
+ assert "WHERE purge_audit_log.id IN" in scope_sql
+ assert all(str(id_) in scope_sql for id_ in ids)
+ assert str(mock_db.session.execute.call_args_list[2].args[0]).startswith("SELECT")
+ delete_sql: str = str(
+ mock_db.session.execute.call_args_list[3]
+ .args[0]
+ .compile(
+ dialect=postgresql.dialect(),
+ compile_kwargs={"literal_binds": True},
+ )
+ )
+ assert delete_sql.startswith("DELETE FROM purge_audit_log")
+ assert str(ids[0]) in delete_sql
+ assert str(ids[1]) not in delete_sql
+ assert "SELECT" not in delete_sql
+
+
+@pytest.mark.parametrize(
+ ("scope_entities", "has_repeat"),
+ [
+ ([("chart", "entity")], True),
+ ([], False),
+ ([("chart", None)], False),
+ ],
+)
+def test_window_repeat_predicate_executes_on_sqlite(
+ scope_entities: list[tuple[str, str | None]],
+ has_repeat: bool,
+) -> None:
+ """Preserve mixed-reason ties and force rows in SQLite window execution."""
+ engine: sa.Engine = sa.create_engine("sqlite://")
+ metadata: sa.MetaData = sa.MetaData()
+ table: sa.Table = prune_audit.PurgeAuditLog.__table__.to_metadata(metadata)
+ ids: list[UUID] = [uuid4() for _ in range(5)]
+ rows: list[dict[str, Any]] = [
+ {
+ "id": id_,
+ "entity_type": "chart",
+ "entity_uuid": "entity",
+ "status": STATUS_BLOCKED,
+ "trigger": trigger,
+ "actor": "system",
+ "created_on": datetime(2026, 1, day),
+ "reason": reason,
+ }
+ for id_, day, reason, trigger in zip(
+ ids,
+ [1, 2, 3, 3, 4],
+ [None, None, None, "changed", "changed"],
+ ["scheduled", "scheduled", "scheduled", "scheduled", "force"],
+ strict=False,
+ )
+ ]
+ try:
+ metadata.create_all(engine)
+ with engine.begin() as connection:
+ connection.execute(sa.insert(table), rows)
+ assert set(
+ connection.scalars(
+ sa.select(table.c.id).where(
+ prune_audit._repeats_an_earlier_block(
+ table, datetime(2026, 2, 1), scope_entities=scope_entities
+ )
+ )
+ )
+ ) == ({ids[1]} if has_repeat else set())
+ finally:
+ engine.dispose()