mikebridge commented on code in PR #45029:
URL: https://github.com/apache/superset/pull/45029#discussion_r4238400334
##########
superset/tasks/deletion_retention.py:
##########
@@ -133,71 +144,178 @@ def _report_model_counts(outcome: str, counts: dict[str,
int]) -> None:
)
-@dataclass
-class _PassTotals:
- """What one pass over the soft-delete roots produced."""
+def _count_eligible(model: type[SoftDeleteMixin], cutoff: datetime) -> int:
+ """Count aged roots, including roots that a later cascade may block."""
+ table: sa.Table = _model_table(model)
+ with skip_visibility_filter(db.session, model):
+ return int(
+ db.session.scalar(
+ sa.select(sa.func.count())
+ .select_from(table)
+ .where(table.c.deleted_at.is_not(None), table.c.deleted_at <
cutoff)
+ )
+ or 0
+ )
- purged: dict[str, int] = field(default_factory=dict)
- would_purge: dict[str, int] = field(default_factory=dict)
- unsupported: dict[str, int] = field(default_factory=dict)
- cascade_failures: int = 0
- blocked: int = 0
+
+class _PurgeScan(NamedTuple):
+ """Counts and supported models collected across one purge invocation."""
+
+ purged: dict[str, int]
+ would_purge: dict[str, int]
+ unsupported_models: dict[str, int]
+ failures: int
+ blocked: int
+ remaining_budget: int | None
+ supported_models: list[type[SoftDeleteMixin]]
scan_failures: int = 0
- #: Roots this pass actually reached, so a run where every one of them
- #: failed can be told apart from a run where one did.
attempted: int = 0
+ commit_uncertain: bool = False
+
+
+class _PurgeModelResult(NamedTuple):
+ """Named per-root counts, including an indeterminate commit attempt."""
+
+ purged: int
+ would_purge: int
+ failures: int
+ blocked: int
+ scan_failures: int
+ commit_uncertain: bool = False
-def _purge_roots(cutoff: datetime, dry_run: bool) -> _PassTotals:
- """Process each registered root, isolating one root's failure from the
rest."""
- totals = _PassTotals()
- for model in _soft_delete_models():
- totals.attempted += 1
- # The table name is itself read off the model, so resolving it belongs
- # inside the guard: a root that cannot supply one would otherwise end
- # the pass here, above the isolation meant to contain it. The class
- # name stands in until it is known, for the failure log.
- entity_type = model.__name__
+class _PurgeCommitUncertainError(RuntimeError):
+ """The database did not acknowledge whether a root purge committed."""
+
+
+def _scan_purge_models(
+ cutoff: datetime, dry_run: bool, max_per_run: int | None
+) -> _PurgeScan:
+ """Apply one shared root budget across every supported model."""
+ purged: dict[str, int] = {}
+ would_purge: dict[str, int] = {}
+ unsupported_models: dict[str, int] = {}
+ failures: int = 0
+ blocked: int = 0
+ scan_failures: int = 0
+ attempted: int = 0
+ commit_uncertain: bool = False
+ remaining_budget: int | None = max_per_run
+ supported_models: list[type[SoftDeleteMixin]] = []
+
+ for model in _ordered_purge_models(cutoff):
+ attempted += 1
+ entity_type: str = model.__name__
try:
entity_type = _model_table_name(model)
if model not in purge_policy_registry():
- totals.unsupported[entity_type] = 1
+ attempted -= 1
+ unsupported_models[entity_type] = 1
logger.warning(
"deletion_retention: skipping %s: no purge policy",
entity_type
)
stats_logger_manager.instance.incr(
f"{_METRIC_PREFIX}.unsupported_models.{entity_type}"
)
continue
- purged_n, would_n, failed_n, blocked_n, scan_failed_n =
_purge_model(
- model, cutoff, dry_run
+ supported_models.append(model)
+ if remaining_budget == 0 and not dry_run:
+ attempted -= 1
+ continue
+ counts: _PurgeModelResult = _purge_model(
+ model,
+ cutoff,
+ dry_run,
+ max_per_run=None if dry_run else remaining_budget,
)
except Exception: # pylint: disable=broad-except
- # One root must not cost the others their run. _purge_model keeps
- # its own counts when a scan fails part-way; this guards what is
- # left -- resolving the registry, and anything else before the
- # first page.
db.session.rollback() # pylint: disable=consider-using-transaction
- totals.scan_failures += 1
+ scan_failures += 1
stats_logger_manager.instance.incr(
f"{_METRIC_PREFIX}.scan_failures.{entity_type}"
)
logger.exception(
"deletion_retention: %s could not be processed", entity_type
)
continue
- if would_n:
- totals.would_purge[entity_type] = would_n
- if purged_n:
- totals.purged[entity_type] = purged_n
- totals.cascade_failures += failed_n
- totals.blocked += blocked_n
- totals.scan_failures += scan_failed_n
- if scan_failed_n:
+ if remaining_budget is not None and not dry_run:
+ # Reserve the possible commit even though it is not confirmed.
+ remaining_budget -= counts.purged + int(counts.commit_uncertain)
+ if counts.would_purge:
+ would_purge[entity_type] = counts.would_purge
+ if counts.purged:
+ purged[entity_type] = counts.purged
+ failures += counts.failures
+ blocked += counts.blocked
+ scan_failures += counts.scan_failures
+ if counts.scan_failures:
stats_logger_manager.instance.incr(
f"{_METRIC_PREFIX}.scan_failures.{entity_type}"
)
- return totals
+ if counts.commit_uncertain:
+ commit_uncertain = True
+ break
+
+ return _PurgeScan(
+ purged=purged,
+ would_purge=would_purge,
+ unsupported_models=unsupported_models,
+ failures=failures,
+ blocked=blocked,
+ remaining_budget=remaining_budget,
+ supported_models=supported_models,
+ scan_failures=scan_failures,
+ attempted=attempted,
+ commit_uncertain=commit_uncertain,
+ )
+
+
+def _add_purge_cap_stats(
+ stats: dict[str, Any],
+ cutoff: datetime,
+ scan: _PurgeScan,
+ max_per_run: int | None,
+) -> None:
+ """Count remaining roots without masking a committed purge on failure."""
+ if max_per_run is None:
+ return
+ remaining_eligible: int | None
+ count_complete: bool = scan.scan_failures == 0 and not
scan.commit_uncertain
+ if count_complete:
+ try:
+ remaining_eligible = sum(
+ _count_eligible(model, cutoff) for model in
scan.supported_models
+ )
+ except Exception: # pylint: disable=broad-except
+ db.session.rollback() # pylint: disable=consider-using-transaction
Review Comment:
Fixed in b7ee9156476ad59778a607209fbd2f3a4b31d259. The remainder-count
handler now catches and logs a failed rollback, then returns the confirmed
purge totals, cap status and unknown-remainder flags.
The existing regression now covers both successful and failed rollback. The
failure case reproduced the escaping exception before the fix; all 114
deletion-retention tests and changed-file hooks pass.
Implemented with Codex; independently reviewed and approved by Claude.
--
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]
