sadpandajoe commented on code in PR #45029:
URL: https://github.com/apache/superset/pull/45029#discussion_r4233928503
##########
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
+ )
+
+
+class _PurgeScan(NamedTuple):
+ """Counts and supported models collected across one purge invocation."""
- 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
+ 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
Review Comment:
If `_purge_model` hits a scan error after it has already committed some
purges, its handler at line 475 calls `db.session.rollback()` before returning
the counts it kept. If that rollback raises (for example the connection is
gone), the exception escapes to this handler, which only counts a scan failure
and `continue`s without subtracting those committed purges from
`remaining_budget`. With a cap of 1,000, a model that has committed 500 roots
and then fails this way leaves the next model a full 1,000, so one run can
delete 1,500 roots while reporting 1,000.
Should an exception that escapes `_purge_model` stop the pass, or should the
model's committed count be kept even when the rollback fails, so the shared
budget stays accurate?
--
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]
