mikebridge commented on code in PR #45029:
URL: https://github.com/apache/superset/pull/45029#discussion_r4200496271
##########
tests/unit_tests/tasks/test_deletion_retention.py:
##########
@@ -573,4 +576,181 @@ def test_standalone_window_bounds_reach_safe_purge_cutoff(
assert result == {"skipped": 1}
purge.assert_not_called()
else:
- purge.assert_called_once_with(Slice, now - timedelta(days=expected),
False)
+ purge.assert_called_once_with(
+ Slice, now - timedelta(days=expected), False, max_per_run=None
+ )
+
+
+def test_purge_cap_counts_committed_roots_across_batches(app_context: None) ->
None:
+ """A failed or blocked root does not consume the successful-purge
budget."""
+ from superset.commands.deletion_retention.purge_cascade import
CascadeResult
+ from superset.models.slice import Slice
+ from superset.tasks import deletion_retention as task
+
+ purged: CascadeResult = CascadeResult(
+ purged=True, entity_type="chart", entity_uuid="purged"
+ )
+ not_purged: CascadeResult = CascadeResult(
+ purged=False, entity_type="chart", entity_uuid="not-purged"
+ )
+ purge_one: MagicMock
+ with (
+ patch.object(task, "_iter_eligible_ids", return_value=[[1, 2], [3,
4]]) as scan,
+ patch.object(
+ task, "_purge_one", side_effect=[not_purged, purged, purged]
+ ) as purge_one,
+ ):
+ result: tuple[int, int, int, int] = task._purge_model(
+ Slice, datetime.now(), dry_run=False, max_per_run=2
+ )
+
+ assert result == (2, 0, 0, 0)
+ assert [call.args[1] for call in purge_one.call_args_list] == [1, 2, 3]
+ scan.assert_called_once()
+
+
+def test_scheduled_purge_rejects_invalid_cap_before_work(app_config: Config)
-> None:
+ """Malformed scheduled budgets fail closed instead of removing
everything."""
+ from superset.tasks import deletion_retention as task
+
+ app_config["SOFT_DELETE_PURGE_MAX_PER_RUN"] = -1
+ purge: MagicMock
+ with (
+ patch.object(task, "_purge_impl") as purge,
+ patch.object(
+ task.feature_flag_manager, "is_feature_enabled", return_value=True
+ ),
+ ):
+ result: dict[str, Any] = task.purge_soft_deleted.run()
+
+ assert result == {"skipped_invalid_cap": 1}
+ purge.assert_not_called()
+
+
+def test_purge_budget_spans_models_and_runs(app_context: None) -> None:
+ """A full first run leaves later models for a subsequent invocation."""
+ from superset.models.dashboard import Dashboard
+ from superset.models.slice import Slice
+ from superset.tasks import deletion_retention as task
+
+ purge: MagicMock
+ with (
+ patch.object(task, "_ordered_purge_models", return_value=[Slice,
Dashboard]),
+ patch.object(
+ task,
+ "_purge_model",
+ side_effect=[
+ (2, 0, 0, 0),
+ (0, 0, 0, 0),
+ (1, 0, 0, 0),
+ ],
+ ) as purge,
+ patch.object(task, "_count_eligible", side_effect=[0, 3, 0, 2]),
+ patch.object(task.audit, "reconcile_pending"),
+ ):
+ first: dict[str, Any] = task._purge_impl(30, False, max_per_run=2)
+ second: dict[str, Any] = task._purge_impl(30, False, max_per_run=2)
+
+ assert first["purged"] == {"slices": 2}
+ assert first["cap_reached"] is True
+ assert first["remaining_eligible"] == 3
+ assert second["purged"] == {"dashboards": 1}
+ assert second["cap_reached"] is False
+ assert second["remaining_eligible"] == 2
+ assert [entry.args[0] for entry in purge.call_args_list] == [
+ Slice,
+ Slice,
+ Dashboard,
+ ]
+ assert [entry.kwargs["max_per_run"] for entry in purge.call_args_list] == [
Review Comment:
Thanks, that was a real gap in the test. I added a case where the first
model purges one root under a cap of two and the second model has two eligible
roots (afb05913b2).
The test runs the real `_purge_model` with the cascade stubbed, so the
second model's budget of one shows up as exactly one purge and a remaining
budget of zero. It fails if the original cap is passed to each model. The
budget code itself was already right; only the test changed.
##########
superset/tasks/version_history_retention.py:
##########
@@ -351,8 +352,46 @@ def _chunked(items: list[int], size: int) ->
Iterator[list[int]]:
_METRIC_PREFIX: str = "superset.versioning.retention"
+def _live_bearing_tables(tables: ShadowTables) -> list[sa.Table]:
+ """Return every shadow table that can anchor a live transaction."""
+ live_bearing_tables: list[sa.Table] = [*tables.parent, *tables.child]
+ if tables.m2m is not None:
+ live_bearing_tables.append(tables.m2m)
+ return live_bearing_tables
+
+
+def _count_prunable(cutoff: datetime, tables: ShadowTables) -> int:
+ """Count the full eligible transaction backlog without writing rows."""
+ total: int = 0
+ after_id: int = 0
+ live_bearing_tables: list[sa.Table] = _live_bearing_tables(tables)
+ conn: sa.engine.Connection
+ with db.engine.connect() as conn:
+ while True:
+ window: _PruneWindow = _resolve_prune_window(
+ conn, cutoff, live_bearing_tables, after_id, _MAX_PRUNE_BATCH
+ )
+ total += len(window.prunable)
+ if window.candidate_count < _MAX_PRUNE_BATCH:
+ return total
+ after_id = window.max_candidate_id
Review Comment:
Good catch. The dry-run backlog count kept one connection, and so one read
transaction, open across the whole scan. afb05913b2 moves the connection
context inside the window loop, so each window's read transaction ends before
the next starts, with a test that checks the first connection closes before the
next window opens. The deleting pass still uses its bounded serializable
transaction.
--
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]
