tdfs4ds.migration.managers

tdfs4ds.migration.managers

Filter / Time manager handling for cross-store migration (v0.3.1.150).

A migrated process may reference a FilterManager (segmented ingestion; the process view's scalar-subquery filter reads its public view) and/or be driven by a TimeManager (roll-out timeline, recorded in FS_TIME_MANAGER). When the process moves to the primary store its managers must move too — but rather than blindly duplicating a manager, we prefer to unify the migrated process onto an already-present, content-equivalent primary manager (avoids manager sprawl), falling back to recreating the manager in the primary when no equivalent exists.

This reuses: * the content-hash equivalence keys from :mod:tdfs4ds.scheduling.parallel_plan (_filter_content_key / _time_content_key), and * the repoint mutators from :mod:tdfs4ds.scheduling.unify (_repoint_process_filter_manager / _repoint_process_time_manager).

Everything is best-effort / fail-open: a manager problem is logged and the process still migrates (its data recomputes; only the segmentation/time driver metadata is affected).

logger_safe

logger_safe(level, message, *args, **kwargs)

Wrapper around the global logger that only emits logs when tdfs4ds.DISPLAY_LOGS is True. level is a string like "info", "error", etc.

manager_content_key

manager_content_key(kind: str, database: Optional[str], view: Optional[str]) -> Optional[str]

Stable content hash of a filter/time manager (None on any problem).

build_primary_manager_index

build_primary_manager_index(kind: str) -> Dict[str, Dict[str, str]]

{content_key: {'database','view','table'}} for every distinct manager in the primary FS_FILTER_MANAGER / FS_TIME_MANAGER catalog.

Used to find a content-equivalent primary manager to unify a migrated process onto. Fail-open to an empty index (then every source manager is recreated).

decide_manager_action

decide_manager_action(kind: str, src_manager: Dict[str, str], primary_index: Dict[str, Dict[str, str]]) -> Dict[str, Any]

Decide whether to unify a migrated process onto an equivalent primary manager or recreate the source manager in the primary.

Parameters

kind : 'filter' | 'time' src_manager : dict {'database','view','table'} of the source manager. primary_index : dict Output of :func:build_primary_manager_index for the same kind.

Returns

dict {'kind', 'source': src_manager, 'decision': 'unify'|'recreate', 'canonical': {...}|None, 'content_key': str|None}.

recreate_manager

recreate_manager(kind: str, src_manager: Dict[str, str], primary_schema: str, dst_view: Optional[str] = None) -> Optional[Dict[str, str]]

Recreate a source manager in the primary store (backing table + view) by replaying its rows through the manager class's own load_* path (so the deterministic id/PI/stats invariants are preserved).

Returns the primary manager descriptor {'database','view','table'} on success, else None (fail-open — the caller keeps the source reference).

repoint_process

repoint_process(kind: str, process_id: str, view_name: str, current: Optional[Dict[str, str]], canonical: Dict[str, str], dry_run: bool = False) -> Dict[str, Any]

Point process_id at canonical manager — rewrites the process view's manager reference and updates the manager catalog row. Wraps the :mod:tdfs4ds.scheduling.unify repoint mutators.