This is an automated email from the ASF dual-hosted git repository.
aminghadersohi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/superset.git
The following commit(s) were added to refs/heads/master by this push:
new afa98f3391a perf(deletion-retention): one window pass for repeat
predicate (#44349)
afa98f3391a is described below
commit afa98f3391aac8022c926b7e07ccebc7d588451c
Author: Amin Ghadersohi <[email protected]>
AuthorDate: Sat Oct 3 18:58:35 2026 +1000
perf(deletion-retention): one window pass for repeat predicate (#44349)
---
.../commands/deletion_retention/prune_audit.py | 76 ++++++++++++++++---
.../deletion_retention/prune_audit_tests.py | 88 ++++++++++++++++++++++
.../deletion_retention/test_prune_audit.py | 33 +++++++-
3 files changed, 186 insertions(+), 11 deletions(-)
diff --git a/superset/commands/deletion_retention/prune_audit.py
b/superset/commands/deletion_retention/prune_audit.py
index 7d14048a693..927f9eb6b90 100644
--- a/superset/commands/deletion_retention/prune_audit.py
+++ b/superset/commands/deletion_retention/prune_audit.py
@@ -347,6 +347,20 @@ def _in_current_streak(target: sa.FromClause, now:
datetime) -> sa.ColumnElement
return sa.or_(boundary.is_(None), target.c.created_on > boundary)
+#: Column names of the timestamp-groups derived table used by the window
+#: repeat check. The groups' labels, the named window's ``ORDER BY`` and the
+#: ``LAG`` columns are all built from these, so renaming one cannot drift the
+#: others: the window text is raw SQL and would otherwise keep the old name.
+_GROUP_TS = "ts"
+_GROUP_LAG_COLUMNS: tuple[str, ...] = (
+ _GROUP_TS,
+ "n",
+ "n_coded",
+ "min_reason",
+ "max_reason",
+)
+
+
def _repeat_path() -> Literal["legacy", "window"]:
"""Select a query using initialized, vendor-normalized server
capabilities."""
dialect: sa.engine.Dialect = db.session.get_bind().dialect
@@ -468,6 +482,31 @@ def _window_repeats_an_earlier_block(
PostgreSQL round-1 plan materialized a groups CTE and joined on entity
alone before filtering ranks, comparing 36 million row pairs.
+ All five ``LAG`` columns share one SQL-level named ``WINDOW w`` (a raw
+ text fragment: SQLAlchemy Core has no construct for a named ``WINDOW``
+ clause) instead of five separate ``LAG(...) OVER (...)`` expressions that
+ happen to repeat the same partition/order spec. PostgreSQL already
+ recognizes five identical inline window specs as one logical pass, but
+ MySQL 8 does not merge them — each materializes its own temporary table,
+ roughly five sequential passes over the batch's timestamp groups. A named
+ window is the SQL-level way to say "this is the same window" so MySQL
+ evaluates it once; PostgreSQL and SQLite (>= 3.25) accept the same syntax
+ unchanged. Only servers routed here run this clause: the MySQL<8,
+ MariaDB<10.2 and SQLite<3.25 fallback in
+ :func:`_legacy_repeats_an_earlier_block` never reaches it.
+
+ An equality join back to a *second* instance of the timestamp-groups
+ derived table (fetching P's aggregates via
+ ``(entity_type, entity_uuid, ts = prev_ts)`` instead of four more ``LAG``
+ columns) was measured and rejected here: PostgreSQL's planner
+ misestimates the derived table's row count for a single-entity batch scope
+ and chooses a Nested Loop over an unindexed ``Materialize`` of the second
+ instance — an O(batch × history) comparison, the same failure shape as
+ the sc-120493 round-1 CTE/entity-only-merge regression, just via a
+ different join path. The named-``WINDOW`` shape keeps the exact join
+ structure already measured safe on PostgreSQL (sc-120493): only the
+ ``LAG`` columns' SQL text changes.
+
Keep the repeat-id query uncorrelated: the sc-120493 Variant 2
measurements showed MySQL repeatedly executing
per-row predecessor scalars. During re-check, scope blocked rows, timestamp
@@ -500,11 +539,12 @@ def _window_repeats_an_earlier_block(
.correlate(None)
.subquery("blocked_rows")
)
+ groups_name: str = "blocked_timestamp_groups"
groups: sa.Subquery = (
sa.select(
source.c.entity_type,
source.c.entity_uuid,
- source.c.created_on.label("ts"),
+ source.c.created_on.label(_GROUP_TS),
sa.func.count().label("n"),
sa.func.count(source.c.reason).label("n_coded"),
sa.func.min(source.c.reason).label("min_reason"),
@@ -514,21 +554,39 @@ def _window_repeats_an_earlier_block(
.where(*blocked_filters, *scope)
.group_by(source.c.entity_type, source.c.entity_uuid,
source.c.created_on)
.correlate(None)
- .subquery("blocked_timestamp_groups")
+ .subquery(groups_name)
+ )
+ # A raw-text named WINDOW clause: the one construct SQLAlchemy Core
+ # cannot emit. ``groups_name`` is the literal alias every LAG column
+ # below must qualify with, so the FROM clause and the WINDOW clause
+ # stay on the same alias. The ordering and LAG column names come from
+ # the module constants shared with the groups' labels.
+ window_body: str = (
+ f"{groups_name}.entity_type, {groups_name}.entity_uuid "
+ f"ORDER BY {groups_name}.{_GROUP_TS}"
)
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")
+ # Raw text carries no type, so copy the grouped column's own
+ # type across. Without it every prev_* lands as NullType and a
+ # later comparison against a Python value would bind it with no
+ # type processor. Types never reach the emitted SQL, which
stays
+ # byte-identical on PostgreSQL, MySQL and SQLite.
+ sa.literal_column(
+ f"lag({groups_name}.{name}) OVER w",
type_=groups.c[name].type
+ ).label(f"prev_{name}")
+ for name in _GROUP_LAG_COLUMNS
],
)
+ .select_from(groups)
+ # Suffixes trail every other clause, so never add ORDER BY/LIMIT to
+ # this select: they would be emitted before WINDOW and fail to parse.
+ # Aliasing this select (``aliased()``/``.alias()``, or any ORM
+ # adaption over ``grp``) rewrites the FROM but not the raw column
+ # text or this suffix, so both would keep naming the old alias.
+ .suffix_with(f"WINDOW w AS (PARTITION BY {window_body})")
.correlate(None)
.subquery("preceding_groups")
)
diff --git a/tests/integration_tests/deletion_retention/prune_audit_tests.py
b/tests/integration_tests/deletion_retention/prune_audit_tests.py
index 5291ee03fc7..05a32400936 100644
--- a/tests/integration_tests/deletion_retention/prune_audit_tests.py
+++ b/tests/integration_tests/deletion_retention/prune_audit_tests.py
@@ -52,6 +52,7 @@ from superset.models.purge_audit_log import (
STATUS_PENDING,
STATUS_TARGET_ABSENT,
)
+from superset.utils import json
from superset.utils.core import backend
from tests.integration_tests.base_tests import SupersetTestCase
from tests.integration_tests.deletion_retention._base import (
@@ -1111,6 +1112,11 @@ class TestRepeatPredicateEquivalence(SupersetTestCase):
now, scope, ids
)
+ # "evidence" is the control: it never reaches the patched
+ # predicate, so equality there asserts isolation from the
+ # rewrite rather than equivalence of it. A change that made it
+ # differ would mean the rewrite had leaked out of the two
+ # categories that call ``_repeats_an_earlier_block``.
for category in ("duplicate", "operational", "evidence"):
assert fallback[category] == legacy[category], (
f"seed={seed} category={category}: legacy fallback
diverged"
@@ -1122,3 +1128,85 @@ class TestRepeatPredicateEquivalence(SupersetTestCase):
)
# Guard against a generator regression that quietly checks nothing.
assert checked_rows > 1000
+
+
+def _windowing_nodes(plan: Any) -> list[dict[str, Any]]:
+ """Every ``windowing`` node in a MySQL ``EXPLAIN FORMAT=JSON`` document."""
+ nodes: list[dict[str, Any]] = []
+ if isinstance(plan, dict):
+ for key, value in plan.items():
+ if key == "windowing":
+ nodes.append(value)
+ nodes.extend(_windowing_nodes(value))
+ elif isinstance(plan, list):
+ for item in plan:
+ nodes.extend(_windowing_nodes(item))
+ return nodes
+
+
+class TestRepeatQueryPlan(SupersetTestCase):
+ """The repeat query's window functions execute as one MySQL pass."""
+
+ 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()
+
+ def test_mysql_repeat_query_evaluates_all_lag_columns_in_one_window_pass(
+ self,
+ ) -> None:
+ """Assert the plan, not just the SQL text.
+
+ MySQL 8 sorts and buffers once per distinct window specification, so
+ five identical inline ``OVER (PARTITION BY ... ORDER BY ...)`` clauses
+ cost five passes over the timestamp groups. The named window must show
+ up in ``EXPLAIN FORMAT=JSON`` as a single windowing step carrying every
+ ``LAG``; the compiled-SQL shape checks alone would not notice the
+ server planning it differently.
+ """
+ dialect: sa.engine.Dialect = db.session.get_bind().dialect
+ if dialect.name != "mysql" or getattr(dialect, "is_mariadb", False):
+ pytest.skip("MySQL-specific execution plan")
+ if prune_audit._repeat_path() != "window":
+ pytest.skip("this server runs the legacy correlated repeat check")
+
+ now: datetime = audit.utc_now()
+ rows: list[dict[str, Any]] = _random_history(random.Random(0), now) #
noqa: S311 — deterministic test data, not crypto
+ db.session.execute(sa.insert(PurgeAuditLog.__table__), rows)
+ db.session.commit()
+
+ table: sa.Table = PurgeAuditLog.__table__
+ query: sa.sql.Select = sa.select(table.c.id).where(
+ table.c.entity_type == _EQUIV_ENTITY_TYPE,
+ *prune_audit._duplicate_predicates(
+ table, now, [(_EQUIV_ENTITY_TYPE, rows[0]["entity_uuid"])]
+ ),
+ )
+ compiled: str = str(
+ query.compile(dialect=dialect, compile_kwargs={"literal_binds":
True})
+ )
+ # The statement is compiled from this module's own constants and
+ # literal-rendered bind values; EXPLAIN takes no bind parameters.
+ plan: dict[str, Any] = json.loads(
+ db.session.execute(sa.text("EXPLAIN FORMAT=JSON " +
compiled)).scalar_one()
+ )
+
+ windowing: list[dict[str, Any]] = _windowing_nodes(plan)
+ assert len(windowing) == 1, "the repeat query must plan one windowing
step"
+ windows: list[dict[str, Any]] = windowing[0]["windows"]
+ assert len(windows) == 1, (
+ "the LAG columns must share one window, not one sort each"
+ )
+ assert windows[0]["functions"] == ["lag"] *
len(prune_audit._GROUP_LAG_COLUMNS)
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 38cb8064179..1b6b53462b3 100644
--- a/tests/unit_tests/commands/deletion_retention/test_prune_audit.py
+++ b/tests/unit_tests/commands/deletion_retention/test_prune_audit.py
@@ -739,7 +739,31 @@ def
test_evidence_drain_skips_scope_lookup_but_rechecks_under_lock() -> None:
] == [row_id]
-def test_repeat_query_uses_lag_without_ctes() -> None:
+def test_only_the_evidence_category_opts_out_of_the_locked_scope_lookup() ->
None:
+ """Wire the opt-out to the one category whose predicates take no scope."""
+ mock_delete: MagicMock
+ with patch.dict(current_app.config, {EVIDENCE_RETENTION_KEY: 3650}):
+ with patch.object(
+ prune_audit, "_delete_batch", return_value=(0, 0)
+ ) as mock_delete:
+ prune_audit.run_prune()
+ # Both entity-partitioned categories must keep the scope lookup: without it
+ # the coordination-locked re-check drops its entity narrowing and rescans
+ # whole history while holding the lock.
+ assert [
+ (type(call.args[1]), call.kwargs["needs_entity_scope"])
+ for call in mock_delete.call_args_list
+ ] == [
+ (prune_audit._DuplicateRecheck, True),
+ (prune_audit._OperationalRecheck, True),
+ (prune_audit._EvidenceRecheck, False),
+ ]
+
+
[email protected](
+ "dialect", [postgresql.dialect(), mysql.dialect(), sqlite.dialect()]
+)
+def test_repeat_query_uses_lag_without_ctes(dialect: Any) -> 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(
@@ -747,10 +771,15 @@ def test_repeat_query_uses_lag_without_ctes() -> None:
table, datetime(2026, 1, 1), scope_entities=[("chart", "entity")]
)
)
- sql: str = str(query.compile(dialect=postgresql.dialect())).lower()
+ sql: str = str(query.compile(dialect=dialect)).lower()
assert "with " not in sql
assert "lag(" in sql
assert "dense_rank(" not in sql
+ # The point of the shape: every LAG shares one named window, so the
+ # engine makes a single ordered pass instead of one per column.
+ assert sql.count("window w as (") == 1
+ assert sql.count("over w") == 5
+ assert "over (partition" not in sql
@pytest.mark.parametrize("count", [1, 2, prune_audit.MAX_BATCH_SIZE])
