sadpandajoe commented on code in PR #44892:
URL: https://github.com/apache/superset/pull/44892#discussion_r4194661969
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -905,32 +1214,407 @@ def _validated_purge_policy(model: type[Any]) ->
PurgeEntityPolicy:
details = (
f"{details}; " if details else ""
) + f"stale_listeners=[{', '.join(coverage.stale_listeners)}]"
- raise RuntimeError(f"Incomplete purge policy for {model.__name__}:
{details}")
+ raise RuntimeError(
+ f"Incomplete purge policy for {policy.model.__name__}: {details}"
+ )
return policy
-def _validate_executable_declarations(policy: PurgeEntityPolicy) -> None:
- """Reject executable classifications missing their required action
metadata."""
+def _ownership_dependency(
+ policy: PurgeEntityPolicy, related_table: str
+) -> DependencyPolicy | None:
+ """The single owned/association edge attaching *related_table*, if
clear."""
+ candidates: tuple[DependencyPolicy, ...] = tuple(
+ dependency
+ for dependency in policy.dependencies
+ if dependency.classification
+ in {DependencyClassification.OWNED,
DependencyClassification.ASSOCIATION}
+ and dependency.key.related_table == related_table
+ and dependency.key.direction == "inbound"
+ )
+ return candidates[0] if len(candidates) == 1 else None
+
+
+def _validate_owned_traversal(policy: PurgeEntityPolicy) -> None:
+ """Reject an owned table reachable only through an association.
+
+ ``cascade_hard_delete`` empties associations before owned children, while
+ an owned table's predicate selects its rows *through* its ownership path
+ (see ``_owner_value_select``). If a hop on that path is an association,
+ its rows are already gone when the owned delete runs: the statement
+ matches nothing, leaving the descendants orphaned where foreign keys are
+ unenforced and blocking the root's delete where they are not.
+
+ Refused at declaration time rather than executed. The shape has a
+ remedy -- classify the intermediate table as owned, which places it in
+ the same phase as what it leads to.
+ """
+ root_table: str = sa.inspect(policy.model).local_table.name
for dependency in policy.dependencies:
- if (
- dependency.classification is
DependencyClassification.LISTENER_EFFECT
- and dependency.listener_action is None
- ):
+ if dependency.classification is not DependencyClassification.OWNED:
+ continue
+ table_name: str = dependency.key.owner_table
+ visited: set[str] = set()
+ while table_name != root_table and table_name not in visited:
+ visited.add(table_name)
+ hop: DependencyPolicy | None = _ownership_dependency(policy,
table_name)
+ if hop is None:
+ # An absent or ambiguous path is reported by coverage and by
+ # _ownership_edge at execution; not this check's business.
+ break
+ if hop.classification is DependencyClassification.ASSOCIATION:
+ raise RuntimeError(
+ f"Owned dependency {dependency.key.describe()} is
reachable "
+ f"only through association {hop.key.describe()}; "
+ "associations are deleted first, so the owned rows would "
+ "be orphaned"
+ )
+ table_name = hop.key.owner_table
+
+
+def _tables_share_a_foreign_key(metadata: sa.MetaData, first: str, second:
str) -> bool:
+ """Whether either table declares a foreign key into the other."""
+ for owner_name, other_name in ((first, second), (second, first)):
+ table: sa.Table | None = metadata.tables.get(owner_name)
+ if table is None:
+ continue
+ for constraint in table.foreign_key_constraints:
+ if constraint.elements[0].column.table.name == other_name:
+ return True
+ return False
+
+
+def _validate_scanner_requirements(policy: PurgeEntityPolicy) -> None:
+ """Reject a root the scheduled scan cannot page through.
+
+ The retention task selects, windows and orders eligible rows by ``id``,
+ and the cascade pins each row by it. A root keyed on something else -- a
+ UUID primary key with no ``id`` column -- raises inside the scan, outside
+ the per-row error handling, so the run aborts before the remaining roots
+ are reached.
+ """
+ mapper: Mapper[Any] = sa.inspect(policy.model)
+ table: sa.Table = mapper.local_table
+ if "id" not in table.c:
+ raise RuntimeError(
+ f"Purge root {policy.model.__name__} has no 'id' column; the "
+ "scheduled scan pages eligible rows by id"
+ )
+ primary_key: tuple[sa.Column[Any], ...] = tuple(mapper.primary_key)
+ if len(primary_key) != 1 or primary_key[0].name != "id":
+ # The scan pages by id and the row is then fetched with a scalar
+ # Session.get, so a composite or differently named key matches
+ # nothing: every row fails while a dry run still counts it eligible.
+ raise RuntimeError(
+ f"Purge root {policy.model.__name__} is keyed on "
+ f"({', '.join(column.name for column in primary_key)}); the "
+ "scheduled purge looks rows up by a single 'id' primary key"
+ )
+ if "deleted_at" not in table.c:
+ raise RuntimeError(
+ f"Purge root {policy.model.__name__} has no 'deleted_at' column; "
+ "every purge path requires the row to be archived first"
+ )
+
+
+def _validate_recursive_ownership(policy: PurgeEntityPolicy) -> None:
+ """Reject a self-referencing owned table under the stock cleanup.
+
+ ``delete_owned_children`` issues one statement per declared edge, so a
+ table that owns itself is pruned one level deep: a three-level tree
+ leaves its deepest rows behind -- orphaned where foreign keys are
+ unenforced, and blocking the root's own delete where they are not. A host
+ declaring this shape supplies cleanup that walks the whole sub-tree.
+ """
+ for dependency in policy.dependencies:
+ if dependency.classification is not DependencyClassification.OWNED:
+ continue
+ key: DependencyKey = dependency.key
+ if key.owner_table != key.related_table:
+ continue
+ if policy.delete_owned_children is delete_owned_children:
raise RuntimeError(
- f"Missing listener action for {dependency.key.describe()}"
+ f"Owned dependency {key.describe()} is self-referencing; the "
+ "stock owned-child cleanup deletes one level, so this policy "
+ "must supply its own delete_owned_children"
)
- if (
- dependency.classification is DependencyClassification.VERSION_OWNED
- and dependency.version_column is None
+
+
+def _validate_sibling_ordering(
+ policy: PurgeEntityPolicy, classification: DependencyClassification
+) -> None:
+ """Reject same-depth siblings of *classification* that reference one
another.
+
+ Owned and association deletes both run through the same depth sort, which
+ is stable -- so two tables at the same depth are deleted in the order they
+ were declared. With a foreign key between them, that order decides the
+ outcome: removing the referenced table first is refused by an enforced
+ constraint and rolls the purge back. Ordering deletes by their own foreign
+ keys would be the richer fix; refusing the shape keeps the engine's
+ contract honest until something needs it.
+
+ Depth is best-effort. A graph the stock traversal cannot measure -- a
+ multi-level tree, say, whose cleanup the host supplies itself -- is left
+ alone rather than rejected for being unmeasurable.
+ """
+ metadata: sa.MetaData = sa.inspect(policy.model).local_table.metadata
+ by_depth: dict[int, list[DependencyPolicy]] = {}
+ for dependency in policy.dependencies:
+ if dependency.classification is not classification:
+ continue
+ try:
+ depth: int = _dependency_owner_depth(policy, dependency.key)
+ except RuntimeError:
+ continue
+ by_depth.setdefault(depth, []).append(dependency)
+ for siblings in by_depth.values():
+ for index, first in enumerate(siblings):
+ for second in siblings[index + 1 :]:
+ if first.key.related_table == second.key.related_table:
+ continue
+ if _tables_share_a_foreign_key(
+ metadata, first.key.related_table, second.key.related_table
+ ):
+ raise RuntimeError(
+ f"{classification.value} dependencies "
+ f"{first.key.describe()} and {second.key.describe()} "
+ "reference each other; these deletes run in "
+ "declaration order, so whether this graph purges "
+ "depends on it"
+ )
+
+
+#: The entity types each stock listener action can act for. Both read
+#: ``policy.entity_type``: tag cleanup maps it to a tag object type, and the
+#: dataset permission name is only built for a dataset. A root outside these
+#: gets a per-row ValueError, or silence, instead of the declared cleanup.
+_LISTENER_ACTION_ENTITY_TYPES: Mapping[ListenerAction, frozenset[str]] = (
+ MappingProxyType(
+ {
+ ListenerAction.DELETE_TAGGED_OBJECTS: frozenset(
+ {"chart", "dashboard", "dataset"}
+ ),
+ ListenerAction.DELETE_DATASET_PERMISSION: frozenset({"dataset"}),
+ }
+ )
+)
+
+
+#: The phase each listener action is dispatched in by the stock callbacks:
+#: tag cleanup runs with the association deletes, permission cleanup once the
+#: entity row is gone and the captured permission name is available.
+_LISTENER_ACTION_PHASES: Mapping[ListenerAction, ExecutionPhase] =
MappingProxyType(
+ {
+ ListenerAction.DELETE_TAGGED_OBJECTS: ExecutionPhase.ASSOCIATIONS,
+ ListenerAction.DELETE_DATASET_PERMISSION: ExecutionPhase.POST_DELETE,
+ }
+)
+
+
+#: Classifications the shared cleanup executes as SQL, through
+#: ``_dependency_predicates`` -- which only knows how to read an inbound
+#: foreign key.
+_EXECUTABLE_CLASSIFICATIONS: frozenset[DependencyClassification] = frozenset(
+ {
+ DependencyClassification.OWNED,
+ DependencyClassification.ASSOCIATION,
+ DependencyClassification.BLOCK,
+ }
+)
+
+
+def _validate_listener_dispatch(
+ policy: PurgeEntityPolicy, dependency: DependencyPolicy
+) -> None:
+ """Reject a listener effect no selected callback would carry out.
+
+ Effects are dispatched from inside two stock callbacks -- association
+ cleanup and permission cleanup -- each for one phase, and each resolving
+ its target from the root's entity type. A declaration in another phase, in
+ a phase whose callback this policy replaces, or for an entity type the
+ action cannot act on, is never executed: the root purges and its declared
+ cleanup does not run.
+ """
+ action: ListenerAction = cast(ListenerAction, dependency.listener_action)
+ expected_phase: ExecutionPhase | None = _LISTENER_ACTION_PHASES.get(action)
+ if expected_phase is None:
+ raise RuntimeError(
+ f"Unsupported listener action for {dependency.key.describe()}"
+ )
+ if dependency.phase is not expected_phase:
+ raise RuntimeError(
+ f"Listener effect {dependency.key.describe()} is declared for "
+ f"{cast(ExecutionPhase, dependency.phase).value}, but "
+ f"{action.value} is dispatched only in {expected_phase.value}"
+ )
+ if expected_phase is ExecutionPhase.ASSOCIATIONS:
+ dispatched = policy.delete_associations is delete_associations
+ callback = "delete_associations"
+ else:
+ dispatched = policy.cleanup_permission is cleanup_dataset_permission
+ callback = "cleanup_permission"
+ if not dispatched:
+ raise RuntimeError(
+ f"Listener effect {dependency.key.describe()} is dispatched by the
"
+ f"stock {callback} callback, which this policy replaces"
+ )
+ supported: frozenset[str] = _LISTENER_ACTION_ENTITY_TYPES[action]
+ if policy.entity_type not in supported:
+ # The stock actions resolve what to delete from the root's entity
+ # type. An unknown one raises per row for tag cleanup, and yields no
+ # permission name at all for permission cleanup -- which the cascade
+ # then skips silently.
+ raise RuntimeError(
+ f"Listener effect {dependency.key.describe()} runs {action.value},
"
+ f"which supports only the entity types "
+ f"{', '.join(sorted(supported))}; this policy declares "
+ f"{policy.entity_type!r}"
+ )
+
+
+def _validate_version_target(
+ policy: PurgeEntityPolicy, dependency: DependencyPolicy, metadata:
sa.MetaData
+) -> None:
+ """Reject a version target whose column does not hold the root's id.
+
+ Version cleanup compares the declared column with the root's id directly,
+ so the column has to be root-relative. Shadow tables are generated and may
+ not exist when a policy is validated, so the check reads their live
+ counterpart -- ``<table>_version`` describes ``<table>``, the naming the
+ shadow declarations already rely on -- and accepts the column when it is
+ the root's own primary key or a foreign key into the root.
+
+ A column naming an intermediate owner, a stage id under a workflow root
+ say, matches history belonging to a different root and leaves this one's
+ behind.
+ """
+ root_table: sa.Table = sa.inspect(policy.model).local_table
+ column: str = cast(str, dependency.version_column)
+ shadow_name: str = dependency.key.related_table
+ suffix: str = "_version"
+ live_name: str = (
+ shadow_name[: -len(suffix)] if shadow_name.endswith(suffix) else
shadow_name
+ )
+ live_table: sa.Table | None = metadata.tables.get(live_name)
+ if live_table is not None and column in live_table.c:
+ live_column: sa.Column[Any] = live_table.c[column]
+ primary_key_names: set[str] = {
+ entry.name for entry in root_table.primary_key.columns
+ }
+ if live_table is root_table and live_column.name in primary_key_names:
+ return
+ if any(
Review Comment:
This accepts a foreign key to any column on the root table, but version
cleanup compares the target column with the root's `id` (`shadow.c[col] ==
entity_id`). With roots `(id=10, code=100)` and `(id=20, code=10)` and an owned
child keyed on `root.code` declared as the version target, purging root 10
deletes root 20's child history and leaves root 10's behind. Should this
require `foreign_key.column` to be the root's single primary-key column?
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -860,16 +896,287 @@ def declare(
cleanup_permission=cleanup_dataset_permission,
),
}
- return validate_unique_root_policies(registry.values())
+ return tuple(registry.values())
-@lru_cache(maxsize=None)
-def _validated_purge_policy(model: type[Any]) -> PurgeEntityPolicy:
- """Validate and return one root policy without blocking unrelated roots."""
+def _policy_label(policy: Any) -> str:
+ """A log-safe name for a payload whose shape is not yet established."""
+ model: Any = getattr(policy, "model", None)
+ return str(getattr(model, "__name__", type(model).__name__))
+
+
+def _host_policy_payload(provided: Any) -> tuple[Any, ...] | None:
+ """Materialize the provider's payload, or reject it.
+
+ Any iterable is accepted -- a list, a tuple, ``dict.values()``, a
+ generator -- since the documented contract is a sequence of policies, not
+ one particular container. Strings and bytes are excluded because they
+ iterate into characters, which would read as a sequence of bad members
+ rather than the wrong type. Materialized once: a generator cannot be
+ walked twice.
+ """
+ if isinstance(provided, (str, bytes)) or not isinstance(provided,
Iterable):
+ logger.error(
+ "purge_policy: %s returned %s; expected an iterable of
PurgeEntityPolicy",
+ HOST_POLICIES_CONFIG_KEY,
+ type(provided).__name__,
+ )
+ return None
try:
- policy: PurgeEntityPolicy = purge_policy_registry()[model]
- except KeyError as ex:
- raise ValueError(f"Unsupported purge model: {model.__name__}") from ex
+ return tuple(provided)
+ except Exception: # pylint: disable=broad-except
+ logger.exception(
+ "purge_policy: %s could not be read; keeping built-in roots only",
+ HOST_POLICIES_CONFIG_KEY,
+ )
+ return None
+
+
+def _unqualified_table_conflict(policy: PurgeEntityPolicy) -> str | None:
+ """Return the first table a bare-name lookup would resolve wrongly.
+
+ Discovery records table identities by bare name while a ``MetaData`` keys
+ them by schema, so a declared ``child`` resolves to the default-schema
+ table even where the policy meant ``host.child`` -- and the cleanup would
+ delete rows belonging to unrelated roots. Carrying schema-qualified
+ identities through discovery and execution is the better fix; until then
+ the ambiguous shape is refused rather than silently mis-resolved.
+ """
+ root_table: sa.Table = sa.inspect(policy.model).local_table
+ metadata: sa.MetaData = root_table.metadata
+ keys_by_name: dict[str, list[str]] = {}
+ for table in metadata.tables.values():
+ keys_by_name.setdefault(table.name, []).append(table.key)
+ declared: set[str] = {root_table.name} | {
+ dependency.key.related_table
+ for dependency in policy.dependencies
+ if dependency.classification in _EXECUTABLE_CLASSIFICATIONS
+ }
+ for name in sorted(declared):
+ keys: list[str] = keys_by_name.get(name, [])
+ if len(keys) != 1 or keys[0] != name:
+ return name
+ return None
+
+
+def _admitted_host_policy(
+ candidate: Any, builtin_roots: frozenset[type[Any]], builtin_types:
frozenset[str]
+) -> PurgeEntityPolicy | None:
+ """Return *candidate* if it is a usable host policy, else ``None``.
+
+ Every check runs behind one handler rather than guarding each attribute:
+ the payload is host-supplied, so reading ``model`` or hashing it may
+ itself raise -- and an exception escaping here would break
+ ``purge_policy_registry`` for the built-in roots too, which is the
+ opposite of what this boundary is for.
+ """
+ try:
+ if not isinstance(candidate, PurgeEntityPolicy):
+ logger.error(
+ "purge_policy: %s returned a %s; expected PurgeEntityPolicy",
+ HOST_POLICIES_CONFIG_KEY,
+ type(candidate).__name__,
+ )
+ return None
+ if not isinstance(candidate.model, type):
+ # Everything downstream treats the root as a mapped class:
+ # sa.inspect, the registry index, the duplicate count.
+ logger.error(
+ "purge_policy: host policy root %s is not a class",
+ _policy_label(candidate),
+ )
+ return None
+ if candidate.model in builtin_roots:
+ logger.error(
+ "purge_policy: host policy for built-in root %s ignored",
+ candidate.model.__name__,
+ )
+ return None
+ if candidate.entity_type in builtin_types:
+ # Core branches on entity_type -- the dataset impact path, the tag
+ # object type, the dataset permission name -- so a host reusing
+ # one of its names would have that logic pointed at its own rows.
+ logger.error(
+ "purge_policy: host policy for %s claims the reserved entity
type %r",
+ candidate.model.__name__,
+ candidate.entity_type,
+ )
+ return None
+ # Validated first, so a table the metadata simply does not contain is
+ # reported as such rather than as an ambiguous name.
+ _validated_policy(candidate)
+ conflict: str | None = _unqualified_table_conflict(candidate)
+ if conflict is not None:
+ raise RuntimeError(
+ f"table {conflict!r} cannot be identified unambiguously by "
+ "name; schema-qualified roots are not supported"
+ )
+ except Exception as ex: # pylint: disable=broad-except
+ # Validated at admission rather than when a row is purged: a
+ # declaration the cleanup cannot execute would otherwise fail once per
+ # eligible row, counted as a cascade failure, and stay invisible until
+ # something aged past the window.
+ logger.error(
+ "purge_policy: host policy for %s rejected: %s",
+ _policy_label(candidate),
+ ex,
+ )
+ return None
+ return candidate
+
+
+def _host_purge_policies(
+ provider: Callable[[], Any] | None,
+) -> tuple[PurgeEntityPolicy, ...]:
+ """Return the purge policies a host installed for its own roots.
+
+ A host distribution can carry ``SoftDeleteMixin`` entities this package
+ cannot import. The retention task discovers those entities through the
+ mixin registry, so without a policy they reach the cascade as an
+ unsupported model. The host therefore declares their purge behavior and
+ installs it under ``PURGE_POLICIES_FUNC``.
+
+ Host boundary: everything about a host policy is settled here, where it is
+ admitted -- an unavailable provider, a payload that is not an iterable of
+ policies, a root that is not a class, a collision with a root or entity
+ type declared in this package, two declarations for one root, and a
+ declaration the shared cleanup could not execute. Each is logged and
+ dropped, so that root is reported as unsupported instead of failing row by
+ row in a scheduled run. A broken host declaration must not take the purge
+ down with it, and must never redefine how a chart, dashboard or dataset is
+ purged.
+ """
+ if provider is None:
+ return ()
+ try:
+ provided: Any = provider()
+ except Exception: # pylint: disable=broad-except
+ logger.exception(
+ "purge_policy: %s is unavailable; keeping built-in roots only",
+ HOST_POLICIES_CONFIG_KEY,
+ )
+ return ()
+ candidates: tuple[Any, ...] | None = _host_policy_payload(provided)
+ if candidates is None:
+ return ()
+
+ builtin_policies: tuple[PurgeEntityPolicy, ...] = _builtin_purge_policies()
+ builtin_roots: frozenset[type[Any]] = frozenset(
+ policy.model for policy in builtin_policies
+ )
+ builtin_types: frozenset[str] = frozenset(
+ policy.entity_type for policy in builtin_policies
+ )
+ admitted: list[PurgeEntityPolicy] = [
+ policy
+ for candidate in candidates
+ if (policy := _admitted_host_policy(candidate, builtin_roots,
builtin_types))
+ is not None
+ ]
+
+ # Counted after admission, where every root is known to be a class and so
+ # hashable. Which of two declarations for one root is authoritative is
+ # undecidable, and the loser would still delete rows, so both are dropped:
+ # the root is reported as unsupported, which is the recoverable outcome.
+ # It also keeps the duplicate away from validate_unique_root_policies,
+ # whose ValueError would abort the whole scheduled run.
+ counts: dict[type[Any], int] = {}
+ for policy in admitted:
+ counts[policy.model] = counts.get(policy.model, 0) + 1
+ for model, count in counts.items():
+ if count > 1:
+ logger.error(
+ "purge_policy: %s declares %d policies for %s; ignoring all of
them",
+ HOST_POLICIES_CONFIG_KEY,
+ count,
+ model.__name__,
+ )
+ return tuple(policy for policy in admitted if counts[policy.model] == 1)
+
+
+#: Sentinel distinguishing "no index resolved yet" from an index resolved for
+#: no provider, which is the ordinary case.
+_UNRESOLVED: Any = object()
+
+
+@dataclass
+class _ResolvedRegistry:
+ """The index built for one provider, with the roots validated so far.
+
+ Compared by provider *identity* rather than memoized with ``lru_cache``: a
+ cache keyed on the provider would hash it, and a host callable
+ implemented as a mutable dataclass instance is unhashable. That would
+ raise here -- before the host boundary could isolate the host's mistake --
+ and stop the built-in roots purging too.
+ """
+
+ provider: Any = _UNRESOLVED
+ registry: Mapping[type[Any], PurgeEntityPolicy] = field(
+ default_factory=lambda: MappingProxyType({})
+ )
+ validated: set[type[Any]] = field(default_factory=set)
+
+
+_RESOLVED: _ResolvedRegistry = _ResolvedRegistry()
+
+#: Serializes rebuilds so concurrent first uses invoke the provider once.
+_RESOLVE_LOCK: threading.Lock = threading.Lock()
+
+
+def _resolved_registry() -> _ResolvedRegistry:
+ """Return the state resolved for the installed provider, building it once.
+
+ Readers take a snapshot and the rebuild publishes a new one, so a caller
+ always sees an index and its validation record together. Rebuilds are
+ serialized and re-checked under the lock: two concurrent first uses would
+ otherwise both invoke the provider, and the slower one would publish over
+ the other -- discarding host roots it had admitted if that second
+ invocation happened to fail transiently.
+ """
+ provider: Callable[[], Any] | None = (
+ current_app.config.get(HOST_POLICIES_CONFIG_KEY) if has_app_context()
else None
+ )
+ resolved: _ResolvedRegistry = _RESOLVED
+ if resolved.provider is provider:
+ return resolved
+ with _RESOLVE_LOCK:
Review Comment:
The provider runs here while the non-reentrant `_RESOLVE_LOCK` is held and
before the new state is published. If a host provider builds its policy from a
built-in one (for example `replace(get_purge_policy(Slice), model=HostRoot,
...)`), `get_purge_policy` re-enters `_resolved_registry()` and blocks forever
on the same lock, stalling the scheduled purge and every later registry lookup.
Should provider construction run outside the lock, or should re-entry be
detected and fail closed?
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -905,32 +1214,407 @@ def _validated_purge_policy(model: type[Any]) ->
PurgeEntityPolicy:
details = (
f"{details}; " if details else ""
) + f"stale_listeners=[{', '.join(coverage.stale_listeners)}]"
- raise RuntimeError(f"Incomplete purge policy for {model.__name__}:
{details}")
+ raise RuntimeError(
+ f"Incomplete purge policy for {policy.model.__name__}: {details}"
+ )
return policy
-def _validate_executable_declarations(policy: PurgeEntityPolicy) -> None:
- """Reject executable classifications missing their required action
metadata."""
+def _ownership_dependency(
+ policy: PurgeEntityPolicy, related_table: str
+) -> DependencyPolicy | None:
+ """The single owned/association edge attaching *related_table*, if
clear."""
+ candidates: tuple[DependencyPolicy, ...] = tuple(
+ dependency
+ for dependency in policy.dependencies
+ if dependency.classification
+ in {DependencyClassification.OWNED,
DependencyClassification.ASSOCIATION}
+ and dependency.key.related_table == related_table
+ and dependency.key.direction == "inbound"
+ )
+ return candidates[0] if len(candidates) == 1 else None
+
+
+def _validate_owned_traversal(policy: PurgeEntityPolicy) -> None:
+ """Reject an owned table reachable only through an association.
+
+ ``cascade_hard_delete`` empties associations before owned children, while
+ an owned table's predicate selects its rows *through* its ownership path
+ (see ``_owner_value_select``). If a hop on that path is an association,
+ its rows are already gone when the owned delete runs: the statement
+ matches nothing, leaving the descendants orphaned where foreign keys are
+ unenforced and blocking the root's delete where they are not.
+
+ Refused at declaration time rather than executed. The shape has a
+ remedy -- classify the intermediate table as owned, which places it in
+ the same phase as what it leads to.
+ """
+ root_table: str = sa.inspect(policy.model).local_table.name
for dependency in policy.dependencies:
- if (
- dependency.classification is
DependencyClassification.LISTENER_EFFECT
- and dependency.listener_action is None
- ):
+ if dependency.classification is not DependencyClassification.OWNED:
+ continue
+ table_name: str = dependency.key.owner_table
+ visited: set[str] = set()
+ while table_name != root_table and table_name not in visited:
+ visited.add(table_name)
+ hop: DependencyPolicy | None = _ownership_dependency(policy,
table_name)
+ if hop is None:
+ # An absent or ambiguous path is reported by coverage and by
+ # _ownership_edge at execution; not this check's business.
+ break
+ if hop.classification is DependencyClassification.ASSOCIATION:
+ raise RuntimeError(
+ f"Owned dependency {dependency.key.describe()} is
reachable "
+ f"only through association {hop.key.describe()}; "
+ "associations are deleted first, so the owned rows would "
+ "be orphaned"
+ )
+ table_name = hop.key.owner_table
+
+
+def _tables_share_a_foreign_key(metadata: sa.MetaData, first: str, second:
str) -> bool:
+ """Whether either table declares a foreign key into the other."""
+ for owner_name, other_name in ((first, second), (second, first)):
+ table: sa.Table | None = metadata.tables.get(owner_name)
+ if table is None:
+ continue
+ for constraint in table.foreign_key_constraints:
+ if constraint.elements[0].column.table.name == other_name:
+ return True
+ return False
+
+
+def _validate_scanner_requirements(policy: PurgeEntityPolicy) -> None:
+ """Reject a root the scheduled scan cannot page through.
+
+ The retention task selects, windows and orders eligible rows by ``id``,
+ and the cascade pins each row by it. A root keyed on something else -- a
+ UUID primary key with no ``id`` column -- raises inside the scan, outside
+ the per-row error handling, so the run aborts before the remaining roots
+ are reached.
+ """
+ mapper: Mapper[Any] = sa.inspect(policy.model)
+ table: sa.Table = mapper.local_table
+ if "id" not in table.c:
+ raise RuntimeError(
+ f"Purge root {policy.model.__name__} has no 'id' column; the "
+ "scheduled scan pages eligible rows by id"
+ )
+ primary_key: tuple[sa.Column[Any], ...] = tuple(mapper.primary_key)
+ if len(primary_key) != 1 or primary_key[0].name != "id":
Review Comment:
A host root with a single UUID primary key named `id` passes this check, but
the scan in `_iter_eligible_ids` starts its watermark at integer `0` and
filters `id > 0`, which fails on PostgreSQL. Its eligible rows are never purged
and every run reports a scan failure. Could admission also require an integer
`id` primary key?
##########
superset/commands/deletion_retention/purge_policy.py:
##########
@@ -905,32 +1214,407 @@ def _validated_purge_policy(model: type[Any]) ->
PurgeEntityPolicy:
details = (
f"{details}; " if details else ""
) + f"stale_listeners=[{', '.join(coverage.stale_listeners)}]"
- raise RuntimeError(f"Incomplete purge policy for {model.__name__}:
{details}")
+ raise RuntimeError(
+ f"Incomplete purge policy for {policy.model.__name__}: {details}"
+ )
return policy
-def _validate_executable_declarations(policy: PurgeEntityPolicy) -> None:
- """Reject executable classifications missing their required action
metadata."""
+def _ownership_dependency(
+ policy: PurgeEntityPolicy, related_table: str
+) -> DependencyPolicy | None:
+ """The single owned/association edge attaching *related_table*, if
clear."""
+ candidates: tuple[DependencyPolicy, ...] = tuple(
+ dependency
+ for dependency in policy.dependencies
+ if dependency.classification
+ in {DependencyClassification.OWNED,
DependencyClassification.ASSOCIATION}
+ and dependency.key.related_table == related_table
+ and dependency.key.direction == "inbound"
+ )
+ return candidates[0] if len(candidates) == 1 else None
+
+
+def _validate_owned_traversal(policy: PurgeEntityPolicy) -> None:
+ """Reject an owned table reachable only through an association.
+
+ ``cascade_hard_delete`` empties associations before owned children, while
+ an owned table's predicate selects its rows *through* its ownership path
+ (see ``_owner_value_select``). If a hop on that path is an association,
+ its rows are already gone when the owned delete runs: the statement
+ matches nothing, leaving the descendants orphaned where foreign keys are
+ unenforced and blocking the root's delete where they are not.
+
+ Refused at declaration time rather than executed. The shape has a
+ remedy -- classify the intermediate table as owned, which places it in
+ the same phase as what it leads to.
+ """
+ root_table: str = sa.inspect(policy.model).local_table.name
for dependency in policy.dependencies:
- if (
- dependency.classification is
DependencyClassification.LISTENER_EFFECT
- and dependency.listener_action is None
- ):
+ if dependency.classification is not DependencyClassification.OWNED:
+ continue
+ table_name: str = dependency.key.owner_table
+ visited: set[str] = set()
+ while table_name != root_table and table_name not in visited:
+ visited.add(table_name)
+ hop: DependencyPolicy | None = _ownership_dependency(policy,
table_name)
+ if hop is None:
Review Comment:
An owned table with two ownership parents that also owns children (root `R`,
owned `A` and `B`, owned `X` reached from both `A` and `B`, owned `Y` under
`X`) is admitted: this loop breaks on the ambiguous hop and the
sibling-ordering check skips what it cannot measure. At execution
`delete_owned_children` raises `Expected one ownership path to x`, so every
eligible row fails with a cascade failure on every run. Should admission reject
an owned table with more than one ownership path instead of deferring to
execution?
--
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]
