#!/usr/bin/env python3 """Two-year disposal evidence (FTC Safeguards Rule, 16 CFR 314.4(c)(6)) from Snowflake, Databricks and Redshift exports, in DuckDB. python3 safeguards_disposal_evidence.py --demo # clean generated records python3 safeguards_disposal_evidence.py --demo --plant # generated records with planted findings python3 safeguards_disposal_evidence.py --demo --plant-input # generated records with planted input faults python3 safeguards_disposal_evidence.py --demo --plant-files # a missing column and a missing file python3 safeguards_disposal_evidence.py --queries # set EXPORT_DIR first: print the row-count query for # every keyed table in YOUR table_register.csv there python3 safeguards_disposal_evidence.py # your exports: set the settings below first python3 safeguards_disposal_evidence.py --strict # exit 1 unless nothing needs follow-up (--work-dir DIR puts the generated/ and evidence-out/ folders somewhere other than the current folder) The rule (314.4(c)(6)(i)): dispose of customer information "no later than two years after the last date the information is used in connection with the provision of a product or service to the customer to which it relates", unless it is necessary for business operations or other legitimate business purposes, is otherwise required to be retained by law or regulation, or targeted disposal is not reasonably feasible. (c)(6)(ii): "Periodically review your data retention policy to minimize the unnecessary retention of data". No platform knows when a customer was last served. You keep that date in your own system of record. You export: customer_register.csv customer_id, last_use_date (YYYY-MM-DD), relationship_open (true or false) table_register.csv platform, table_name, customer_key_column (blank if the table has no customer key) customer_rows.csv platform, table_name, customer_id, row_count, measured_on (--queries prints the query) exceptions.csv kind (customer or table), id, exception, basis, ends_on, review_by review_log.csv review_date, reviewed_by plus each platform's own recovery settings (Snowflake ACCOUNT_USAGE.TABLES, Databricks table properties and DESCRIBE HISTORY, Redshift describe-cluster-snapshots). It prints five lists, each with its filters, row count and a SHA-256 of its rows, and writes each as a CSV: 1. customers past two years that still have rows (PAST TWO YEARS, LAST USE UNCLEAR, NOT IN REGISTER, EXCEPTION IN FORCE), and 1b the tables read, with NOT MEASURED where a table could not be read by customer 2. exceptions, with NO BASIS, EXPIRED and NOT REVIEWED flags (the script never judges a stated business need) 3. disposed but still recoverable: days a row deleted on AS_OF can still be brought back, per table 4. retention policy reviews on record and the days between them (the rule gives no interval: set one) The script reads files; it never connects to a platform and never writes to one. The demo records are GENERATED (seed 234): not a company's data. Column names come from: Snowflake https://docs.snowflake.com/en/sql-reference/account-usage/tables Databricks https://docs.databricks.com/aws/en/tables/history-schema https://docs.databricks.com/aws/en/sql/language-manual/delta-vacuum Redshift https://docs.aws.amazon.com/cli/latest/reference/redshift/describe-cluster-snapshots.html Our assumptions, to check against one of your own rows: Databricks DESCRIBE HISTORY names a finished VACUUM "VACUUM END" and a purge REORG (the history schema page does not list the operation names), so both are settings. """ import argparse import csv import datetime as dt import hashlib import pathlib import random import re import sys import duckdb # ---- CHANGE THESE FOR YOUR OWN DATA --------------------------------------------------------- AS_OF = "2026-11-01" # the day you took the exports (UTC) EXPORT_DIR = None # folder holding the export files named in FILES below PLATFORMS_EXPORTED = ["snowflake", "databricks", "redshift"] # a platform not listed is NOT CHECKED in list 3 RETENTION_REVIEW_DAYS = None # no default: the rule says "periodically". Your longest allowed gap, in days REDSHIFT_AUTOMATED_RETENTION_DAYS = None # no default: your cluster's automated snapshot retention in days DATABRICKS_DEFAULT_RETENTION_DAYS = 7 # delta.deletedFileRetentionDuration when the table does not set it DATABRICKS_VACUUM_DONE = "VACUUM END" # the DESCRIBE HISTORY operation that is a finished VACUUM DATABRICKS_PURGE_OPS = ["REORG"] # the operation that rewrites files with deletion vectors (REORG ... APPLY (PURGE)) SNOWFLAKE_MIN_DATA_RETENTION_DAYS = 0 # your account's MIN_DATA_RETENTION_TIME_IN_DAYS, 0 if not set ROW_COUNTS_MAX_STALE_DAYS = 3 # row counts must have been measured within this many days of AS_OF # ---------------------------------------------------------------------------------------------- FAILSAFE_DAYS = 7 # Snowflake: "a (non-configurable) 7-day period" (docs.snowflake.com/en/user-guide/data-failsafe) FILES = { "customer_register": "customer_register.csv", "table_register": "table_register.csv", "customer_rows": "customer_rows.csv", "exceptions": "exceptions.csv", "review_log": "review_log.csv", "snowflake_tables": "snowflake_tables.csv", "databricks_table_properties": "databricks_table_properties.csv", "databricks_history": "databricks_history.csv", "redshift_snapshots": "redshift_snapshots.csv", } PLATFORM_FILES = { "snowflake": ["snowflake_tables"], "databricks": ["databricks_table_properties", "databricks_history"], "redshift": ["redshift_snapshots"], } COLUMNS = { # the platforms' published column names, lower-cased here; the file's own case does not matter "customer_register": ["customer_id", "last_use_date", "relationship_open"], "table_register": ["platform", "table_name", "customer_key_column"], "customer_rows": ["platform", "table_name", "customer_id", "row_count", "measured_on"], "exceptions": ["kind", "id", "exception", "basis", "ends_on", "review_by"], "review_log": ["review_date", "reviewed_by"], "snowflake_tables": ["table_catalog", "table_schema", "table_name", "is_transient", "retention_time", "deleted"], "databricks_table_properties": ["table_name", "deleted_file_retention_days", "deletion_vectors"], "databricks_history": ["table_name", "timestamp", "operation"], "redshift_snapshots": ["snapshotidentifier", "snapshotcreatetime", "snapshottype", "manualsnapshotretentionperiod"], } PLATFORMS = ["snowflake", "databricks", "redshift"] # The rule's three exceptions, in short forms taken from 314.4(c)(6)(i); exceptions.csv must use these exact forms. EXCEPTIONS = ["business operations or other legitimate business purposes", "required by law or regulation", "targeted disposal is not reasonably feasible"] OUT_DIR = pathlib.Path("evidence-out") GEN_DIR = pathlib.Path("generated") DATE_RE = re.compile(r"\d{4}-\d{2}-\d{2}") TS_RE = re.compile(r"(\d{4}-\d{2}-\d{2})([ T]\d{2}:\d{2}:\d{2}(\.\d+)?( ?(Z|\+00:00|\+0000))?)?") IDENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_$]*$") PART = r'(?:"(?:[^"]|"")+"|[A-Za-z0-9_$]+)' NAME = re.compile(rf"^{PART}(\.{PART})*$") # dotted parts, each plain or double-quoted def pdate(s): """A date read strictly: YYYY-MM-DD only. 8/5/26 and 2026/09/24 are None, and counted by a check.""" s = (s or "").strip() if not DATE_RE.fullmatch(s): return None try: return dt.date.fromisoformat(s) except ValueError: return None def pts(s): """The UTC date of a timestamp written YYYY-MM-DD HH:MM:SS (a T, fraction, Z or +00:00 is accepted).""" m = TS_RE.fullmatch((s or "").strip()) return pdate(m.group(1)) if m else None def dn(n): return f"{n} DAY" if int(n) == 1 else f"{int(n)} DAYS" def norm_table(s): return (s or "").replace('"', "").strip().lower() def quote_name(platform, table): """The table name in the platform's own quoting: a double-quoted part stays quoted (backticks on Databricks).""" parts = re.findall(PART, table) if platform == "databricks": parts = ["`" + p[1:-1].replace('""', '"').replace("`", "``") + "`" if p.startswith('"') else p for p in parts] return ".".join(parts) def query_for(platform, table, key): cast = "STRING" if platform == "databricks" else "VARCHAR" lit = norm_table(table).replace("'", "''") return (f"SELECT '{platform}' AS platform, '{lit}' AS table_name, TRIM(CAST({key} AS {cast})) AS customer_id,\n" f" COUNT(*) AS row_count, CURRENT_DATE AS measured_on\nFROM {quote_name(platform, table)}\nGROUP BY 1, 2, 3;") # --------------------------------------------------------------------------------------------- # generated demo records (seed 234) # --------------------------------------------------------------------------------------------- def write_csv(path, header, rows): with open(path, "w", newline="", encoding="utf-8") as f: w = csv.writer(f) w.writerow(header) w.writerows(rows) def generate(gen_dir, plant, plant_input, plant_files=False): gen_dir.mkdir(parents=True, exist_ok=True) rnd = random.Random(234) asof = dt.date.fromisoformat(AS_OF) tables = [("snowflake", "sf_prod.crm.customers", "customer_id"), ("snowflake", "sf_prod.loans.applications", "cust_id"), ("snowflake", "sf_prod.ops.app_logs", ""), ("databricks", "main.crm.customers", "customer_id"), ("databricks", "main.loans.payments", "borrower_id"), ("redshift", "crm.customers", "customer_id"), ("redshift", "loans.applications", "applicant_id")] keyed = [t for t in tables if t[2]] cust, rows, exc = [], [], [] def add_rows(cid, n_tables): picks = keyed[:] if n_tables is None else rnd.sample(keyed, n_tables) for p, t, _ in sorted(picks): rows.append([p, t, cid, str(rnd.randint(1, 40))]) for i in range(1, 51): # served in the last two years cid = f"C-{100000 + i}" cust.append([cid, (asof - dt.timedelta(days=rnd.randint(10, 700))).isoformat(), "true" if rnd.random() < .8 else "false"]) add_rows(cid, rnd.randint(2, 4)) for i in range(51, 57): # served more than two years ago and already disposed: no rows cust.append([f"C-{100000 + i}", (asof - dt.timedelta(days=rnd.randint(760, 1500))).isoformat(), "false"]) held = [("C-100057", 900, EXCEPTIONS[1], "Retention schedule item 4 (document LR-17)", "", "2027-03-31"), ("C-100058", 1100, EXCEPTIONS[0], "Open dispute file DF-22, owner: legal", "", "2027-01-15")] for cid, ago, typ, basis, ends, review in held: cust.append([cid, (asof - dt.timedelta(days=ago)).isoformat(), "false"]) add_rows(cid, 2) exc.append(["customer", cid, typ, basis, ends, review]) cust.append(["C-100059", "2024-11-01", "false"]) # two years end today: due, not yet past add_rows("C-100059", 2) exc.append(["table", "snowflake:sf_prod.ops.app_logs", EXCEPTIONS[2], "Log lines hold customer ids inside free text and cannot be found one customer at a time", "", "2027-02-28"]) sf = [["SF_PROD", "CRM", "CUSTOMERS", "NO", "14", ""], ["SF_PROD", "LOANS", "APPLICATIONS", "NO", "30", ""], ["SF_PROD", "OPS", "APP_LOGS", "YES", "1", ""]] dbp = [["main.crm.customers", "", "false"], ["main.loans.payments", "30", "false"]] dbh = [["main.crm.customers", "2026-10-17 03:00:12", "VACUUM END"], ["main.crm.customers", "2026-10-24 03:00:09", "VACUUM END"], ["main.crm.customers", "2026-10-26 11:14:40", "DELETE"], ["main.loans.payments", "2026-10-30 03:01:55", "VACUUM END"]] snaps = [["rs-auto-0001", "2026-10-31 02:00:00", "automated", ""], ["rs-manual-q3", "2026-10-15 00:00:00", "manual", "45"]] reviews = [["2024-05-15", "Head of Data Governance"], ["2025-04-02", "Head of Data Governance"], ["2026-03-09", "Head of Data Governance"]] cust_rows = rows if plant: planted = [("C-100060", "2023-05-04", "false", 3), ("C-100061", "2022-12-19", "false", 2), ("C-100062", "2024-10-31", "false", 2), ("C-100063", "2024-02-29", "false", 1), ("C-100064", "2023-01-15", "true", 2), ("C-100065", "2023-08-20", "false", 2), ("C-100066", "2023-09-02", "false", 1), ("C-100067", "2023-10-10", "false", 2)] for cid, last, op, n in planted: cust.append([cid, last, op]) add_rows(cid, n) exc.append(["customer", "C-100065", EXCEPTIONS[0], "Contract dispute CD-31", "2026-06-30", "2026-12-01"]) exc.append(["customer", "C-100066", EXCEPTIONS[1], "", "", "2027-01-01"]) exc.append(["customer", "C-100067", EXCEPTIONS[1], "Retention schedule item 9", "", "2026-09-30"]) exc = [e for e in exc if e[0] != "table"] # no exception for the log table cust_rows = [r for r in rows if r[1] != "main.loans.payments"] # that table was never counted cust_rows += [["redshift", "loans.applications", "C-100099", "4"], ["snowflake", "sf_prod.crm.customers", "", "12"]] dbh = [r for r in dbh if not (r[0] == "main.loans.payments" and r[2] == "VACUUM END")] + \ [["main.loans.payments", "2026-10-28 09:00:00", "DELETE"]] snaps.append(["rs-manual-2024-close", "2024-12-31 00:00:00", "manual", "-1"]) dbp[0][2] = "true" # deletion vectors on, never purged reviews = [["2024-05-15", "Head of Data Governance"], ["2025-08-14", "Head of Data Governance"]] cr = [r + [AS_OF] for r in cust_rows] files = {"customer_register": (COLUMNS["customer_register"], cust), "table_register": (COLUMNS["table_register"], [list(t) for t in tables]), "customer_rows": (COLUMNS["customer_rows"], cr), "exceptions": (COLUMNS["exceptions"], exc), "review_log": (COLUMNS["review_log"], reviews), "snowflake_tables": (["TABLE_CATALOG", "TABLE_SCHEMA", "TABLE_NAME", "IS_TRANSIENT", "RETENTION_TIME", "DELETED"], sf), "databricks_table_properties": (COLUMNS["databricks_table_properties"], dbp), "databricks_history": (COLUMNS["databricks_history"], dbh), "redshift_snapshots": (["SnapshotIdentifier", "SnapshotCreateTime", "SnapshotType", "ManualSnapshotRetentionPeriod"], snaps)} if plant_input: files["review_log"][1].append(["2026-12-25", "Head of Data Governance"]) # a review dated after AS_OF files["customer_register"][1].extend([list(cust[0]), ["C-100070", "8/5/26", "false"], ["C-100071", "2027-01-10", "false"], ["C-100072", "2025-01-10", "maybe"]]) # duplicate id, bad date, future date, bad flag files["table_register"][1].extend([list(tables[0]), ["snowflak", "sf_prod.x.y", "id"], ["redshift", "crm.keyed_oddly", "cust id"]]) # duplicate, unknown platform, unusable key name cr2 = files["customer_rows"][1] cr2.extend([["redshift", "crm.old_staging", "C-100001", "5", AS_OF], ["redshift", "crm.customers", "C-100002", "12x", AS_OF], list(cr2[0]), ["snowflake", "sf_prod.ops.app_logs", "C-100003", "2", AS_OF]]) # not in register, bad count, duplicate, keyless table cr2[1][4] = "2026-09-01" # counted too long ago files["exceptions"][1].extend([["group", "C-100001", EXCEPTIONS[1], "x", "", "2027-01-01"], ["customer", "C-100002", "we just like it", "x", "", "2027-01-01"], ["customer", "C-999999", EXCEPTIONS[1], "x", "", "2027-01-01"], ["customer", "C-100003", EXCEPTIONS[1], "x", "31/12/2026", "2027-01-01"]]) files["snowflake_tables"][1][1][4] = "one week" files["databricks_table_properties"][1][1][1] = "7d" files["databricks_table_properties"][1][0][2] = "maybe" files["databricks_history"][1].extend([["main.crm.customers", "2026-10-28 03:00:00", "VACUUM FULL"], ["main.crm.customers", "28 Oct 2026 3am", "DELETE"]]) files["redshift_snapshots"][1].extend([["rs-manual-bad", "last Tuesday", "manual", "30"], ["rs-manual-month", "2026-10-20 00:00:00", "manual", "a month"]]) if plant_files: files["review_log"] = (["review_date"], [[r[0]] for r in reviews]) # no reviewed_by column del files["snowflake_tables"] # a file that is not there for name, (header, data) in files.items(): write_csv(gen_dir / FILES[name], header, data) # --------------------------------------------------------------------------------------------- def main(): global EXPORT_DIR, RETENTION_REVIEW_DAYS, REDSHIFT_AUTOMATED_RETENTION_DAYS, SNOWFLAKE_MIN_DATA_RETENTION_DAYS ap = argparse.ArgumentParser() ap.add_argument("--demo", action="store_true") ap.add_argument("--plant", action="store_true") ap.add_argument("--plant-input", action="store_true") ap.add_argument("--plant-files", action="store_true") ap.add_argument("--queries", action="store_true") ap.add_argument("--strict", action="store_true") ap.add_argument("--work-dir", default=".") a = ap.parse_args() work = pathlib.Path(a.work_dir) asof = pdate(AS_OF) if asof is None: sys.exit("AS_OF must be written YYYY-MM-DD") con = duckdb.connect() if a.demo: GEN_DIR_ = work / GEN_DIR generate(GEN_DIR_, a.plant, a.plant_input, a.plant_files) EXPORT_DIR = GEN_DIR_ # Demo settings: ours, to let the lists run. The rule sets neither interval. RETENTION_REVIEW_DAYS = 365 REDSHIFT_AUTOMATED_RETENTION_DAYS = None if a.plant_input else 1 if a.plant_input: SNOWFLAKE_MIN_DATA_RETENTION_DAYS = -1 # a setting that cannot be right print("Demo records are GENERATED (seed 234); not a company's data.") print(f"Demo settings: AS_OF {AS_OF}, RETENTION_REVIEW_DAYS {RETENTION_REVIEW_DAYS}, " f"REDSHIFT_AUTOMATED_RETENTION_DAYS {REDSHIFT_AUTOMATED_RETENTION_DAYS}") if EXPORT_DIR is None: if a.queries: sys.exit("Set EXPORT_DIR to the folder holding table_register.csv, then run --queries.") sys.exit("Set EXPORT_DIR at the top of the script, or run with --demo.") exp = pathlib.Path(EXPORT_DIR) file_problems = [] def load(name): path = exp / FILES[name] if not path.is_file(): file_problems.append(f"{FILES[name]}: not found") return [] try: cur = con.execute("SELECT * FROM read_csv(?, header=true, all_varchar=true, null_padding=true)", [str(path)]) head = [d[0].strip().lower() for d in cur.description] data = cur.fetchall() except duckdb.Error: file_problems.append(f"{FILES[name]}: cannot be read as a CSV file") return [] missing = [c for c in COLUMNS[name] if c not in head] if missing: file_problems.append(f"{FILES[name]}: missing column {', '.join(missing)}") return [] out = [] for i, r in enumerate(data, 1): d = {h: ("" if v is None else str(v).strip()) for h, v in zip(head, r)} d["_n"] = i out.append(d) return out if a.queries: reg = load("table_register") if file_problems: sys.exit("; ".join(file_problems)) for r in reg: if not r["customer_key_column"]: continue if r["platform"].lower() not in PLATFORMS or not NAME.match(r["table_name"]) or not IDENT.match(r["customer_key_column"]): sys.exit(f"table_register.csv data row {r['_n']}: platform, table or key column cannot be used in a query") print(query_for(r["platform"].lower(), r["table_name"], r["customer_key_column"]) + "\n") return 0 # ---- load ------------------------------------------------------------------------------- custs = load("customer_register") treg = load("table_register") crow = load("customer_rows") excs = load("exceptions") revs = load("review_log") pdata = {n: load(n) for p in PLATFORMS if p in PLATFORMS_EXPORTED for n in PLATFORM_FILES[p]} checks = [] def chk(label, bad): checks.append((label, bad)) chk("input files are present, readable and have every column", file_problems) # customer register seen, dup = set(), [] for r in custs: if r["customer_id"] in seen: dup.append(r["_n"]) seen.add(r["customer_id"]) chk("customer_register: no customer_id twice or blank", dup + [r["_n"] for r in custs if not r["customer_id"]]) chk("customer_register: last_use_date is YYYY-MM-DD or blank", [r["_n"] for r in custs if r["last_use_date"] and pdate(r["last_use_date"]) is None]) chk("customer_register: no last_use_date after AS_OF", [r["_n"] for r in custs if pdate(r["last_use_date"]) and pdate(r["last_use_date"]) > asof]) chk("customer_register: relationship_open is true or false", [r["_n"] for r in custs if r["relationship_open"].lower() not in ("true", "false")]) # table register keys, dupt = set(), [] for r in treg: k = (r["platform"].lower(), norm_table(r["table_name"])) if k in keys: dupt.append(r["_n"]) keys.add(k) chk("table_register: no table twice", dupt) chk("table_register: platform is snowflake, databricks or redshift", [r["_n"] for r in treg if r["platform"].lower() not in PLATFORMS]) chk("table_register: customer_key_column is a plain column name or blank", [r["_n"] for r in treg if r["customer_key_column"] and not IDENT.match(r["customer_key_column"])]) key_of = {(r["platform"].lower(), norm_table(r["table_name"])): r["customer_key_column"] for r in treg} # customer rows def rk(r): return (r["platform"].lower(), norm_table(r["table_name"])) chk("customer_rows: every table is in table_register", [r["_n"] for r in crow if rk(r) not in key_of]) chk("customer_rows: row_count is a whole number, 0 or more", [r["_n"] for r in crow if not re.fullmatch(r"\d+", r["row_count"])]) seen, dup = set(), [] for r in crow: k = rk(r) + (r["customer_id"],) if k in seen: dup.append(r["_n"]) seen.add(k) chk("customer_rows: no customer counted twice for one table", dup) def stale(r): d = pdate(r["measured_on"]) return d is None or d > asof or (asof - d).days > ROW_COUNTS_MAX_STALE_DAYS chk(f"customer_rows: measured_on is YYYY-MM-DD, not after AS_OF, within {ROW_COUNTS_MAX_STALE_DAYS} days of it", [r["_n"] for r in crow if stale(r)]) chk("customer_rows: no counts for a table the register says has no customer key", [r["_n"] for r in crow if key_of.get(rk(r)) == "" and rk(r) in key_of]) # exceptions known_ids = {r["customer_id"] for r in custs} | {r["customer_id"] for r in crow if r["customer_id"]} def exc_target(e): if e["kind"].lower() == "customer": return ("customer", e["id"]) if e["kind"].lower() == "table" and ":" in e["id"]: p, t = e["id"].split(":", 1) return ("table", (p.strip().lower(), norm_table(t))) return None chk("exceptions: kind is customer or table", [r["_n"] for r in excs if r["kind"].lower() not in ("customer", "table")]) chk("exceptions: exception is one of the rule's three", [r["_n"] for r in excs if r["exception"].lower() not in EXCEPTIONS]) chk("exceptions: every id is a customer in the register or the rows, or a table in table_register", [r["_n"] for r in excs if exc_target(r) is None or (exc_target(r)[0] == "customer" and exc_target(r)[1] not in known_ids) or (exc_target(r)[0] == "table" and exc_target(r)[1] not in key_of)]) chk("exceptions: ends_on and review_by are YYYY-MM-DD or blank", [r["_n"] for r in excs if (r["ends_on"] and pdate(r["ends_on"]) is None) or (r["review_by"] and pdate(r["review_by"]) is None)]) # review log chk("review_log: review_date is YYYY-MM-DD and not after AS_OF", [r["_n"] for r in revs if pdate(r["review_date"]) is None or pdate(r["review_date"]) > asof]) # platform records if "snowflake" in PLATFORMS_EXPORTED: # only the live rows of tables in the register: a view or a dropped table in the export is not a fault sf_reg = {t for p, t in key_of if p == "snowflake"} chk("snowflake_tables: for each register table's live row, RETENTION_TIME is a whole number and IS_TRANSIENT " "is YES or NO", [r["_n"] for r in pdata.get("snowflake_tables", []) if not r["deleted"] and norm_table(f"{r['table_catalog']}.{r['table_schema']}.{r['table_name']}") in sf_reg and (not re.fullmatch(r"\d+", r["retention_time"]) or r["is_transient"].upper() not in ("YES", "NO"))]) chk("SNOWFLAKE_MIN_DATA_RETENTION_DAYS is a whole number, 0 or more", [] if isinstance(SNOWFLAKE_MIN_DATA_RETENTION_DAYS, int) and SNOWFLAKE_MIN_DATA_RETENTION_DAYS >= 0 else ["setting"]) if "databricks" in PLATFORMS_EXPORTED: chk("databricks_table_properties: deleted_file_retention_days is a whole number or blank", [r["_n"] for r in pdata.get("databricks_table_properties", []) if r["deleted_file_retention_days"] and not re.fullmatch(r"\d+", r["deleted_file_retention_days"])]) chk("databricks_table_properties: deletion_vectors is true or false", [r["_n"] for r in pdata.get("databricks_table_properties", []) if r["deletion_vectors"].lower() not in ("true", "false")]) chk(f"databricks_history: every VACUUM operation is {DATABRICKS_VACUUM_DONE} or VACUUM START", [r["_n"] for r in pdata.get("databricks_history", []) if r["operation"].upper().startswith("VACUUM") and r["operation"].upper() not in (DATABRICKS_VACUUM_DONE, "VACUUM START")]) chk("databricks_history: timestamp is YYYY-MM-DD HH:MM:SS", [r["_n"] for r in pdata.get("databricks_history", []) if pts(r["timestamp"]) is None]) if "redshift" in PLATFORMS_EXPORTED: snaps = pdata.get("redshift_snapshots", []) chk("redshift_snapshots: SnapshotCreateTime is YYYY-MM-DD HH:MM:SS and the type is manual or automated", [r["_n"] for r in snaps if pts(r["snapshotcreatetime"]) is None or r["snapshottype"].lower() not in ("manual", "automated")]) chk("redshift_snapshots: a manual snapshot's ManualSnapshotRetentionPeriod is a whole number or -1", [r["_n"] for r in snaps if r["snapshottype"].lower() == "manual" and not re.fullmatch(r"-?\d+", r["manualsnapshotretentionperiod"])]) chk("redshift_snapshots: REDSHIFT_AUTOMATED_RETENTION_DAYS is set when automated snapshots exist", ["setting is None"] if REDSHIFT_AUTOMATED_RETENTION_DAYS is None and any(r["snapshottype"].lower() == "automated" for r in snaps) else []) failed = [(l, b) for l, b in checks if b] print(f"\n== Input checks ({len(checks)} run, {len(failed)} failed)") for label, bad in checks: if bad: shown = ", ".join(str(x) for x in bad[:8]) + (" ..." if len(bad) > 8 else "") print(f"[FAIL] {label}: {'data rows' if isinstance(bad[0], int) else 'detail'} {shown}") else: print(f"[ok] {label}") incomplete = bool(failed) flagged = len(failed) # ---- usable records ---------------------------------------------------------------------- con.execute("CREATE TABLE cust(customer_id VARCHAR, last_use DATE, rel_open BOOLEAN)") seen = set() for r in custs: if r["customer_id"] and r["customer_id"] not in seen: seen.add(r["customer_id"]) con.execute("INSERT INTO cust VALUES (?,?,?)", [r["customer_id"], pdate(r["last_use_date"]), {"true": True, "false": False}.get(r["relationship_open"].lower())]) con.execute("CREATE TABLE rws(tkey VARCHAR, customer_id VARCHAR, n BIGINT)") seen, nokey = set(), {} for r in crow: k = rk(r) if k not in key_of or not re.fullmatch(r"\d+", r["row_count"]) or k + (r["customer_id"],) in seen: continue seen.add(k + (r["customer_id"],)) if not r["customer_id"]: nokey[k] = nokey.get(k, 0) + int(r["row_count"]) elif int(r["row_count"]) > 0: con.execute("INSERT INTO rws VALUES (?,?,?)", [f"{k[0]}:{k[1]}", r["customer_id"], int(r["row_count"])]) # exceptions: in force only when the rule's reason is one of three, a basis is written, it has not ended # and its review date has not passed exc_rows, c_exc, t_exc = [], {}, {} for e in sorted(excs, key=lambda x: (x["kind"].lower(), x["id"], x["exception"], x["_n"])): tgt = exc_target(e) if tgt is None or e["exception"].lower() not in EXCEPTIONS: continue ends, review = pdate(e["ends_on"]), pdate(e["review_by"]) flags = [] if not e["basis"]: flags.append("NO BASIS") if e["ends_on"] and ends is None: flags.append("END DATE UNREADABLE") if ends and ends < asof: flags.append("EXPIRED") if review is None or review < asof: flags.append("NOT REVIEWED") exc_rows.append([tgt, e, flags]) if not flags: (c_exc if tgt[0] == "customer" else t_exc).setdefault(tgt[1], e["exception"].lower()) con.execute("CREATE TABLE cexc(customer_id VARCHAR, exc VARCHAR)") for k, v in c_exc.items(): con.execute("INSERT INTO cexc VALUES (?,?)", [k, v]) con.execute("CREATE TABLE texc(tkey VARCHAR, exc VARCHAR)") for k, v in t_exc.items(): con.execute("INSERT INTO texc VALUES (?,?)", [f"{k[0]}:{k[1]}", v]) # ---- list 1: customers past two years --------------------------------------------------- sql = """ SELECT r.customer_id, c.last_use, c.rel_open, c.customer_id IS NOT NULL AS in_register, c.last_use + INTERVAL 2 YEAR AS two_years, SUM(r.n) AS rows_total, SUM(CASE WHEN te.tkey IS NULL THEN r.n ELSE 0 END) AS rows_not_excepted, string_agg(r.tkey || '=' || CAST(r.n AS VARCHAR), '; ' ORDER BY r.tkey) AS tables, ANY_VALUE(ce.exc) AS cust_exc FROM rws r LEFT JOIN cust c ON c.customer_id = r.customer_id LEFT JOIN texc te ON te.tkey = r.tkey LEFT JOIN cexc ce ON ce.customer_id = r.customer_id GROUP BY r.customer_id, c.last_use, c.rel_open, c.customer_id, two_years ORDER BY r.customer_id""" res = con.execute(sql).fetchall() l1, within = [], 0 for cid, last, op, inreg, two, tot, notex, tbls, cexc_ in res: two = two.date() if two is not None else None if not inreg: status = "NOT IN REGISTER" elif last is None: status = "LAST USE UNCLEAR" elif asof <= two: within += 1 continue elif op is not False: status = "LAST USE UNCLEAR" # an open relationship with an old last use is never cleared elif cexc_ or notex == 0: status = "EXCEPTION IN FORCE" else: status = "PAST TWO YEARS" if cexc_: note = cexc_ elif notex == 0: note = "every table has a table exception" else: note = "" l1.append([cid, status, str(last or ""), str(two or ""), str(tot), str(notex), tbls, note]) n_reg = len({r["customer_id"] for r in custs if r["customer_id"]}) with_rows = len(res) print(f"\nPopulation: {n_reg} customers in the register; {with_rows} customers have rows in a register table; " f"{within} of them are inside two years as of {AS_OF}; {with_rows - within} are listed in 1.") def emit(title, fname, header, rows, filters, flag_col=None, ok_status=()): """Print and write one list; returns how many rows need follow-up (status not starting with an ok word).""" rows = sorted(rows) digest = hashlib.sha256("\n".join("\x1f".join(r) for r in rows).encode("utf-8")).hexdigest() tag = " [INCOMPLETE: an input check failed]" if incomplete else "" print(f"\n== {title}: {fname}{tag}") print(f" as of {AS_OF}; {filters}") print(f" rows: {len(rows)} sha256: {digest}") print(" " + " | ".join(header)) for r in rows: print(" " + " | ".join(r)) out_dir.mkdir(parents=True, exist_ok=True) write_csv(out_dir / f"{fname}.csv", header, rows) n = 0 if flag_col is not None: n = sum(1 for r in rows if not r[flag_col].startswith(ok_status)) return n stamp = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%d-%H%M%S") out_dir = work / OUT_DIR / f"run-{stamp}" k = 1 while out_dir.exists(): k += 1 out_dir = work / OUT_DIR / f"run-{stamp}-{k}" flagged += emit("Customers past two years", "ftc_1_customers_past_two_years", ["customer_id", "status", "last_use_date", "two_years_after", "rows", "rows_not_excepted", "tables", "exception"], l1, "customers with at least one row in a register table, more than two years after the last date of use", flag_col=1, ok_status=("EXCEPTION IN FORCE",)) # ---- list 1b: tables read by customer ------------------------------------------------------- l1b = [] totals = {} for tkey, n in con.execute("SELECT tkey, SUM(n) FROM rws GROUP BY tkey").fetchall(): totals[tkey] = n ncust = dict(con.execute("SELECT tkey, COUNT(DISTINCT customer_id) FROM rws GROUP BY tkey").fetchall()) for (p, t), key in sorted(key_of.items()): tk = f"{p}:{t}" if key == "": if (p, t) in t_exc: status = f"NO CUSTOMER KEY, EXCEPTION IN FORCE ({t_exc[(p, t)]})" else: status = "NOT MEASURED: no customer key and no exception in force" elif (p, t) not in {rk(r) for r in crow}: status = "NOT MEASURED: no row counts for this table" else: status = "MEASURED" if nokey.get((p, t)): status += f"; {nokey[(p, t)]} ROWS WITH NO CUSTOMER KEY" l1b.append([p, t, key, status, str(ncust.get(tk, 0)), str(totals.get(tk, 0))]) flagged += emit("Tables read by customer", "ftc_1b_tables_by_customer", ["platform", "table_name", "customer_key_column", "status", "customers_with_rows", "rows"], l1b, "every table in table_register", flag_col=3, ok_status=("MEASURED", "NO CUSTOMER KEY, EXCEPTION IN FORCE")) # a MEASURED table that also has keyless rows still needs follow-up flagged += sum(1 for r in l1b if r[3].startswith("MEASURED;")) # ---- list 2: exceptions --------------------------------------------------------------------- rows_by_c = dict(con.execute("SELECT customer_id, SUM(n) FROM rws GROUP BY customer_id").fetchall()) l2 = [] for tgt, e, flags in exc_rows: if tgt[0] == "customer": held, ident = rows_by_c.get(tgt[1], 0), tgt[1] else: ident = f"{tgt[1][0]}:{tgt[1][1]}" held = totals.get(ident, 0) if key_of.get(tgt[1]) else None # a table with no key is not counted by customer l2.append([tgt[0], ident, e["exception"].lower(), "yes" if e["basis"] else "no", e["ends_on"], e["review_by"], ", ".join(flags) if flags else "IN FORCE", "not counted" if held is None else str(held)]) flagged += emit("Exceptions", "ftc_2_exceptions", ["kind", "id", "exception", "basis_written", "ends_on", "review_by", "status", "rows_held"], l2, "every exception in exceptions.csv with one of the rule's three reasons; the script never judges the stated need", flag_col=6, ok_status=("IN FORCE",)) # ---- list 3: disposed but still recoverable -------------------------------------------------- sf = {} for r in pdata.get("snowflake_tables", []): if not r["deleted"]: sf.setdefault(norm_table(f"{r['table_catalog']}.{r['table_schema']}.{r['table_name']}"), r) dbp = {norm_table(r["table_name"]): r for r in pdata.get("databricks_table_properties", [])} vac, purge = {}, {} for r in pdata.get("databricks_history", []): if r["operation"].upper() == DATABRICKS_VACUUM_DONE and pts(r["timestamp"]): vac.setdefault(norm_table(r["table_name"]), []).append(pts(r["timestamp"])) if r["operation"].upper() in [o.upper() for o in DATABRICKS_PURGE_OPS] and pts(r["timestamp"]): purge.setdefault(norm_table(r["table_name"]), []).append(pts(r["timestamp"])) # Redshift: what a row deleted on AS_OF can still be brought back from rs_days, rs_indef, rs_basis = 0, False, [] for r in pdata.get("redshift_snapshots", []): created = pts(r["snapshotcreatetime"]) if created is None or created > asof: continue if r["snapshottype"].lower() == "manual" and re.fullmatch(r"-?\d+", r["manualsnapshotretentionperiod"]): n = int(r["manualsnapshotretentionperiod"]) if n == -1: rs_indef = True rs_basis.append("a manual snapshot with retention -1 is kept indefinitely") elif (created + dt.timedelta(days=n) - asof).days > 0: left = (created + dt.timedelta(days=n) - asof).days rs_days = max(rs_days, left) rs_basis.append(f"a manual snapshot has {dn(left).lower()} left") elif r["snapshottype"].lower() == "automated" and REDSHIFT_AUTOMATED_RETENTION_DAYS is not None: rs_days = max(rs_days, int(REDSHIFT_AUTOMATED_RETENTION_DAYS)) rs_basis.append(f"automated snapshots up to {dn(REDSHIFT_AUTOMATED_RETENTION_DAYS).lower()}") rs_basis = sorted(set(rs_basis)) l3 = [] for (p, t) in sorted(key_of): if p not in PLATFORMS_EXPORTED: l3.append([p, t, "", "NOT CHECKED", "platform not in PLATFORMS_EXPORTED"]) elif p == "snowflake": r = sf.get(t) if r is None or not re.fullmatch(r"\d+", r["retention_time"]): l3.append([p, t, "", "NOT MEASURED", "no usable live row in snowflake_tables.csv"]) elif r["is_transient"].upper() == "YES": l3.append([p, t, r["retention_time"], dn(r["retention_time"]), f"transient: Time Travel {dn(r['retention_time']).lower()}, no Fail-safe"]) else: tt = max(int(r["retention_time"]), SNOWFLAKE_MIN_DATA_RETENTION_DAYS if isinstance( SNOWFLAKE_MIN_DATA_RETENTION_DAYS, int) else 0) raised = " (raised by MIN_DATA_RETENTION_TIME_IN_DAYS)" if tt > int(r["retention_time"]) else "" n = tt + FAILSAFE_DAYS l3.append([p, t, str(n), dn(n), f"Time Travel {dn(tt).lower()}{raised} + {dn(FAILSAFE_DAYS).lower()} Fail-safe"]) elif p == "databricks": r = dbp.get(t) if r is None or (r["deleted_file_retention_days"] and not re.fullmatch(r"\d+", r["deleted_file_retention_days"])) \ or r["deletion_vectors"].lower() not in ("true", "false"): l3.append([p, t, "", "NOT MEASURED", "no usable row in databricks_table_properties.csv"]) continue n = int(r["deleted_file_retention_days"] or DATABRICKS_DEFAULT_RETENTION_DAYS) last = max(vac[t]) if t in vac else None if r["deletion_vectors"].lower() == "true": lp = max(purge[t]) if t in purge else None if lp is None: l3.append([p, t, "", "DELETION VECTORS, NO PURGE IN HISTORY", "a DELETE only marks rows: they stay in the table's files until REORG TABLE ... APPLY (PURGE), " f"then at least {dn(n).lower()} until a VACUUM; history is kept 30 days by default"]) else: l3.append([p, t, "", f"UNTIL THE NEXT PURGE, THEN AT LEAST {dn(n)}", f"deletion vectors: rows stay until REORG TABLE ... APPLY (PURGE) (last {lp}), then a VACUUM"]) elif last is None: l3.append([p, t, str(n), f"NO VACUUM IN HISTORY, AT LEAST {dn(n)}", "old files stay until a VACUUM removes them; history is kept 30 days by default, so an older " "VACUUM may not show"]) else: l3.append([p, t, str(n), f"AT LEAST {dn(n)}", f"until the first VACUUM after that; last VACUUM {last} ({dn((asof - last).days).lower()} before AS_OF)"]) else: if rs_indef: l3.append([p, t, "", "INDEFINITE", "; ".join(rs_basis)]) elif rs_basis: l3.append([p, t, str(rs_days), f"UP TO {dn(rs_days)}", "; ".join(rs_basis)]) else: l3.append([p, t, "0", "NO SNAPSHOT IN FORCE", "no snapshot in the export holds a row deleted on AS_OF"]) flagged += emit("Disposed but still recoverable", "ftc_3_recoverable_after_disposal", ["platform", "table_name", "recoverable_days", "status", "basis"], l3, "days a row deleted on AS_OF can still be brought back; a count to look at, not a finding, except " "INDEFINITE, DELETION VECTORS WITH NO PURGE and NOT MEASURED", flag_col=3, ok_status=("INDEFINITE-NEVER",)) # only INDEFINITE, DELETION VECTORS WITH NO PURGE and NOT MEASURED are findings here; the day counts are for reading flagged -= sum(1 for r in l3 if not r[3].startswith(("INDEFINITE", "NOT MEASURED", "DELETION VECTORS, NO PURGE"))) # ---- list 4: retention policy reviews ----------------------------------------------------------- l4 = [] print() if RETENTION_REVIEW_DAYS is None: print("== Retention policy reviews: ftc_4_retention_policy_reviews NOT RUN: RETENTION_REVIEW_DAYS is not set") out_dir.mkdir(parents=True, exist_ok=True) write_csv(out_dir / "ftc_4_retention_policy_reviews.csv", ["date", "days_since_previous", "reviewer_named", "status"], [[AS_OF, "", "", "NOT RUN: RETENTION_REVIEW_DAYS is not set"]]) flagged += 1 else: dates = sorted((pdate(r["review_date"]), r["reviewed_by"]) for r in revs if pdate(r["review_date"]) and pdate(r["review_date"]) <= asof) prev = None for d, who in dates: gap = (d - prev).days if prev else None flags = [] if gap is not None and gap > RETENTION_REVIEW_DAYS: flags.append(f"GAP OVER {RETENTION_REVIEW_DAYS} DAYS") if not who: flags.append("NO REVIEWER") l4.append([str(d), "" if gap is None else str(gap), "yes" if who else "no", ", ".join(flags) or "OK"]) prev = d if not dates: l4.append([AS_OF, "", "", "NO REVIEW ON RECORD"]) else: since = (asof - dates[-1][0]).days l4.append([AS_OF, str(since), "", f"OVERDUE: {since} DAYS SINCE THE LAST REVIEW" if since > RETENTION_REVIEW_DAYS else f"OK: {since} DAYS SINCE THE LAST REVIEW"]) flagged += emit("Retention policy reviews", "ftc_4_retention_policy_reviews", ["date", "days_since_previous", "reviewer_named", "status"], l4, f"reviews on record; the allowed gap is your setting, {RETENTION_REVIEW_DAYS} days; the last row is AS_OF itself", flag_col=3, ok_status=("OK",)) print(f"\nWrote the lists to {out_dir.relative_to(work)}/ (CSV, one file per list).") print(f"Result: {'NEEDS FOLLOW-UP' if flagged else 'nothing needs follow-up'} ({flagged} items)") return 1 if (a.strict and flagged) else 0 if __name__ == "__main__": sys.exit(main())