mikebridge commented on code in PR #45029: URL: https://github.com/apache/superset/pull/45029#discussion_r4209207237
########## 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 + while True: + with db.engine.connect() as conn: + 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 + + +def _probe_prunable(cutoff: datetime, tables: ShadowTables) -> tuple[int, bool]: + """Count one candidate window and say whether it covers the backlog.""" + conn: sa.engine.Connection + with db.engine.connect() as conn: + window: _PruneWindow = _resolve_prune_window( + conn, cutoff, _live_bearing_tables(tables), 0, _MAX_PRUNE_BATCH Review Comment: Agreed. The bounded remainder probe now starts at the scan cursor from the last live pass, which keeps the partially processed last window in view while skipping earlier preserved-anchor windows. I added a regression with a full preserved window followed by a capped prune and a remaining eligible transaction; it fails with the old zero result and passes with the cursor fix. Fixed in https://github.com/apache/superset/commit/4ae52014ad8dea65664f4554070ffc1d056ba1aa. ########## superset/tasks/version_history_retention.py: ########## @@ -556,7 +683,23 @@ def prune_old_versions() -> dict[str, Any]: retention_days: int = _resolve_version_history_retention_days( configured, legacy_configured, seed=_version_history_retention_seed ) - return _prune_old_versions_impl(retention_days) + try: + max_per_run: int | None = validate_retention_cap( + current_app.config.get( + "VERSION_HISTORY_PRUNE_MAX_TRANSACTIONS_PER_RUN", 1000 + ), + "VERSION_HISTORY_PRUNE_MAX_TRANSACTIONS_PER_RUN", + ) + except ValueError: + logger.warning("version_history_retention: invalid prune cap; skipping") + stats_logger_manager.instance.incr(f"{_METRIC_PREFIX}.skipped_invalid_cap") + return {"skipped_invalid_cap": 1} + dry_run: bool = current_app.config.get("VERSION_HISTORY_PRUNE_DRY_RUN", False) + if type(dry_run) is not bool: + raise ValueError("VERSION_HISTORY_PRUNE_DRY_RUN must be a bool") Review Comment: Agreed. A non-boolean dry-run value now logs a warning, increments `skipped_invalid_dry_run` once, and returns without invoking the prune implementation. Six invalid types are covered; they previously returned `{"error": 1}` and incremented the failure counter, while boolean values continue through the existing path. Fixed in https://github.com/apache/superset/commit/4ae52014ad8dea65664f4554070ffc1d056ba1aa. ########## tests/unit_tests/tasks/test_version_history_retention.py: ########## @@ -407,9 +409,260 @@ def test_immediate_cutoff_and_invalid_skip(stats: MagicMock, days: int) -> None: days ) if days == -1: - run_pass.assert_called_once_with(now, [], 0) + run_pass.assert_called_once_with(now, [], 0, max_prune=None) assert result["cutoff"] == now.isoformat() else: assert result == {"skipped": 1} tables.assert_not_called() run_pass.assert_not_called() + + +def test_prune_cap_accounts_only_for_committed_passes(stats: MagicMock) -> None: + """A retried pass uses the same allowance and cannot overshoot the cap.""" + tables: version_history_retention.ShadowTables = ( + version_history_retention.ShadowTables( + parent=[], child=[], m2m=None, transaction=MagicMock() + ) + ) + first: dict[str, int] = { + "candidate_count": 1000, + "max_candidate_id": 1000, + "pruned_transactions": 2, + } + second: dict[str, int] = { + "candidate_count": 1000, + "max_candidate_id": 2000, + "pruned_transactions": 1, + } + run_pass: MagicMock + with ( + patch.object( + version_history_retention, "_resolve_shadow_tables", return_value=tables + ), + patch.object( + version_history_retention, + "_run_pass_with_retry", + side_effect=[(first, 1), (second, 0)], + ) as run_pass, + patch.object( + version_history_retention, "_probe_prunable", return_value=(4, False) + ), + ): + result: dict[str, Any] = version_history_retention._prune_old_versions_impl( + retention_days=30, max_per_run=3 + ) + + assert result["pruned_transactions"] == 3 + assert result["retried"] == 1 + assert result["cap_reached"] is True + assert result["remaining_eligible"] == 4 + assert result["remaining_count_complete"] is False + stats.gauge.assert_any_call( + "superset.versioning.retention.remaining_count_complete", 0 + ) + assert run_pass.call_args_list[0].kwargs["max_prune"] == 3 + assert run_pass.call_args_list[1].kwargs["max_prune"] == 1 + + [email protected]("cap", [None, 0, 3]) +def test_prune_dry_run_counts_all_eligible_without_writes( + stats: MagicMock, cap: int | None +) -> None: + """A dry run reports the whole backlog and capped-run estimate.""" + tables: version_history_retention.ShadowTables = ( + version_history_retention.ShadowTables( + parent=[], child=[], m2m=None, transaction=MagicMock() + ) + ) + run_pass: MagicMock + with ( + patch.object( + version_history_retention, "_resolve_shadow_tables", return_value=tables + ), + patch.object(version_history_retention, "_count_prunable", return_value=7), + patch.object(version_history_retention, "_run_pass_with_retry") as run_pass, + ): + result: dict[str, Any] = version_history_retention._prune_old_versions_impl( + 30, max_per_run=cap, dry_run=True + ) + + assert result["eligible_backlog"] == 7 + assert result["estimated_capped_runs"] == (3 if cap == 3 else 1) + run_pass.assert_not_called() + stats.gauge.assert_not_called() + + [email protected]("helper_name", ["_count_prunable", "_probe_prunable"]) +def test_remainder_reads_use_default_isolation(helper_name: str) -> None: + """Backlog measurement does not request the delete pass's isolation.""" + tables: version_history_retention.ShadowTables = ( + version_history_retention.ShadowTables( + parent=[], child=[], m2m=None, transaction=MagicMock() + ) + ) + engine: MagicMock = MagicMock() + window: version_history_retention._PruneWindow = ( + version_history_retention._PruneWindow( + prunable=[], candidate_count=0, max_candidate_id=0 + ) + ) + with ( + patch.object(version_history_retention, "db") as mock_db, + patch.object( + version_history_retention, "_resolve_prune_window", return_value=window + ), + ): + mock_db.engine = engine + result: int | tuple[int, bool] = getattr( + version_history_retention, helper_name + )(datetime(2026, 1, 1), tables) + + assert result in (0, (0, True)) + engine.connect.return_value.execution_options.assert_not_called() + + +def test_dry_run_count_closes_read_transaction_between_windows() -> None: + """A full backlog scan releases each read snapshot before the next window.""" + tables: version_history_retention.ShadowTables = ( + version_history_retention.ShadowTables( + parent=[], child=[], m2m=None, transaction=MagicMock() + ) + ) + engine: MagicMock = MagicMock() + full_window: version_history_retention._PruneWindow = ( + version_history_retention._PruneWindow( + prunable=[], Review Comment: Agreed. The test now uses two prunable transactions in the first full window and one in the final window, asserts the total is three, and verifies the second resolver call starts after the first window's maximum candidate ID. I confirmed it fails both when the count resets per window and when the cursor stops advancing, while retaining the connection-closure check. Fixed in https://github.com/apache/superset/commit/844caa4c937db3731cdbf7e66048280a020154ba. ########## docs/docs/using-superset/recently-archived.mdx: ########## @@ -159,6 +159,20 @@ destructive when it runs. Set it to `True` in `superset_config.py` and restart the relevant processes to log what the task *would* purge without deleting anything. Validate the eligible backlog before enabling live purging or shortening a window; the first live run can process accumulated old archives. +`SOFT_DELETE_PURGE_MAX_PER_RUN` defaults to `1000` root charts, dashboards, or +datasets per scheduled run. One root's complete cascade counts as one and is +never split to fit the cap. Set `0` or `None` in `superset_config.py` to remove +the cap. It bounds committed deletions, not candidate evaluations: blocked roots +are re-evaluated every run, so a blocked-heavy backlog can still require +substantial work. A negative, boolean, string, or fractional value stops the +scheduled task without purging. A live run reports `cap_reached` and `remaining_eligible` Review Comment: Agreed. The guide now says the cap and remainder fields and gauges appear only for a positive-cap live run. Unlimited mode omits them, so monitoring should not treat their absence after disabling the cap as a measurement failure. Fixed in https://github.com/apache/superset/commit/844caa4c937db3731cdbf7e66048280a020154ba. -- 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]
