sadpandajoe commented on code in PR #45029:
URL: https://github.com/apache/superset/pull/45029#discussion_r4235974885
##########
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:
If the remainder count fails and this `db.session.rollback()` also raises
(for example the connection is gone), the exception escapes `_purge_impl` after
the roots are already committed. The scheduled wrapper then returns `{"error":
1}` and none of the confirmed `purged` totals, `cap_reached` or the
unknown-remainder flags are reported, even though up to the cap was deleted.
`_purge_model` now logs a failed scan rollback and keeps its counts; should
this handler do the same, and should
`test_purge_remainder_failure_keeps_committed_totals` also cover a rollback
that raises?
--
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]
