diff --git a/deploy/airflow/mask_pii_setup.py b/deploy/airflow/mask_pii_setup.py
new file mode 100644
index 0000000..9cb0b6e
--- /dev/null
+++ b/deploy/airflow/mask_pii_setup.py
@@ -0,0 +1,138 @@
+#!/usr/bin/env python3
+"""Add realistic PII columns to the source tables and backfill a bounded sample.
+
+Runs inside the atc-agents api container (has psycopg2/pymysql/pymongo + network
+to 10.0.21.51). Idempotent: ADD COLUMN IF NOT EXISTS / duplicate-column tolerant.
+Backfills the most recent N rows so OpenMetadata auto-classification samples real
+PII values. New generator rows also get PII (see updated generator scripts).
+"""
+import os
+import random
+
+import psycopg2
+import pymysql
+import pymongo
+
+DB_HOST = os.getenv("SRC_DB_HOST", "10.0.21.51")
+DB_USER = os.getenv("SRC_DB_USER", "mo")
+DB_PASS = os.getenv("SRC_DB_PASSWORD", "Dell2026!")
+N = int(os.getenv("PII_BACKFILL_ROWS", "4000"))
+
+FIRST = ["Sophie", "Liam", "Emma", "Noah", "Julia", "Lucas", "Mila", "Daan", "Anna", "Sem",
+         "Eva", "Finn", "Tess", "Bram", "Lotte", "Max", "Sara", "Thijs", "Nina", "Ruben"]
+LAST = ["de Vries", "Jansen", "Bakker", "Visser", "Smit", "Meijer", "Mulder", "Bos",
+        "Vos", "Peters", "Hendriks", "Dijkstra", "Kok", "Willems", "Maas", "Koning"]
+STREETS = ["Kerkstraat", "Dorpsweg", "Molenpad", "Stationsplein", "Industrieweg", "Lindelaan",
+           "Beukenhof", "Havenstraat", "Parkweg", "Schoolstraat"]
+CITIES = ["Amsterdam", "Rotterdam", "Utrecht", "Eindhoven", "Den Haag", "Groningen", "Tilburg", "Almere"]
+DOMAINS = ["example.com", "mail.nl", "acme-corp.eu", "shopmail.com", "contact.io"]
+
+
+def full_name():
+    return f"{random.choice(FIRST)} {random.choice(LAST)}"
+
+
+def email(name):
+    base = name.lower().replace(" ", ".")
+    return f"{base}{random.randint(1, 999)}@{random.choice(DOMAINS)}"
+
+
+def phone():
+    return f"+31 6 {random.randint(10, 99)} {random.randint(100000, 999999)}"
+
+
+def ip():
+    return ".".join(str(random.randint(1, 254)) for _ in range(4))
+
+
+def iban():
+    return f"NL{random.randint(10, 99)}DELL{random.randint(10**9, 10**10 - 1)}"
+
+
+def national_id():
+    return str(random.randint(100000000, 999999999))
+
+
+def address():
+    return f"{random.choice(STREETS)} {random.randint(1, 320)}, {random.randint(1000, 9999)} {random.choice(CITIES)}"
+
+
+def dob():
+    return f"{random.randint(1960, 2002)}-{random.randint(1, 12):02d}-{random.randint(1, 28):02d}"
+
+
+def do_postgres():
+    print("== PostgreSQL sales_orders ==")
+    conn = psycopg2.connect(host=DB_HOST, dbname="postgres", user=DB_USER, password=DB_PASS, port=5432, connect_timeout=10)
+    conn.autocommit = True
+    cur = conn.cursor()
+    for col, typ in [("customer_name", "text"), ("customer_email", "text"), ("customer_phone", "text"),
+                     ("customer_ip", "text"), ("billing_iban", "text"), ("shipping_address", "text")]:
+        cur.execute(f"ALTER TABLE public.sales_orders ADD COLUMN IF NOT EXISTS {col} {typ}")
+    print("  columns ensured")
+    cur.execute("SELECT order_id FROM public.sales_orders ORDER BY order_id DESC LIMIT %s", (N,))
+    ids = [r[0] for r in cur.fetchall()]
+    rows = []
+    for oid in ids:
+        nm = full_name()
+        rows.append((nm, email(nm), phone(), ip(), iban(), address(), oid))
+    cur.executemany(
+        "UPDATE public.sales_orders SET customer_name=%s, customer_email=%s, customer_phone=%s, "
+        "customer_ip=%s, billing_iban=%s, shipping_address=%s WHERE order_id=%s", rows)
+    print(f"  backfilled {len(rows)} rows with PII")
+    cur.close()
+    conn.close()
+
+
+def do_mysql():
+    print("== MySQL employee_events ==")
+    conn = pymysql.connect(host=DB_HOST, user=DB_USER, password=DB_PASS, database="hr", port=3306,
+                           connect_timeout=10, autocommit=True)
+    cur = conn.cursor()
+    for col, typ in [("employee_name", "varchar(120)"), ("employee_email", "varchar(160)"),
+                     ("employee_phone", "varchar(40)"), ("national_id", "varchar(20)"),
+                     ("home_address", "varchar(255)"), ("date_of_birth", "date")]:
+        try:
+            cur.execute(f"ALTER TABLE employee_events ADD COLUMN {col} {typ}")
+        except Exception as e:
+            if "Duplicate column" not in str(e):
+                raise
+    print("  columns ensured")
+    cur.execute("SELECT event_id FROM employee_events ORDER BY event_id DESC LIMIT %s", (N,))
+    ids = [r[0] for r in cur.fetchall()]
+    rows = []
+    for eid in ids:
+        nm = full_name()
+        rows.append((nm, email(nm), phone(), national_id(), address(), dob(), eid))
+    cur.executemany(
+        "UPDATE employee_events SET employee_name=%s, employee_email=%s, employee_phone=%s, "
+        "national_id=%s, home_address=%s, date_of_birth=%s WHERE event_id=%s", rows)
+    print(f"  backfilled {len(rows)} rows with PII")
+    cur.close()
+    conn.close()
+
+
+def do_mongo():
+    print("== MongoDB events ==")
+    uri = os.getenv("SRC_MONGO_URI", f"mongodb://{DB_HOST}:27017/?replicaSet=rs0")
+    client = pymongo.MongoClient(uri, serverSelectionTimeoutMS=10000)
+    coll = client["supplychain"]["events"]
+    docs = list(coll.find({}, {"_id": 1}).sort("_id", -1).limit(N))
+    n = 0
+    for d in docs:
+        nm = full_name()
+        coll.update_one({"_id": d["_id"]}, {"$set": {
+            "contact_name": nm, "contact_email": email(nm), "contact_phone": phone(),
+        }})
+        n += 1
+    print(f"  updated {n} docs with PII fields")
+    client.close()
+
+
+if __name__ == "__main__":
+    for fn in (do_postgres, do_mysql, do_mongo):
+        try:
+            fn()
+        except Exception as e:
+            print(f"  ERROR in {fn.__name__}: {e}")
+    print("done")
diff --git a/deploy/airflow/scripts/generate_mongodb_events_data.py b/deploy/airflow/scripts/generate_mongodb_events_data.py
new file mode 100644
index 0000000..13e42f4
--- /dev/null
+++ b/deploy/airflow/scripts/generate_mongodb_events_data.py
@@ -0,0 +1,78 @@
+#!/usr/bin/env python3
+"""
+Generate fake event data for MongoDB.
+Light + configurable via GEN_ROWS. Now also embeds contact PII fields
+(contact_name/email/phone) for masking + PII classification demos.
+"""
+
+import pymongo
+import random
+from datetime import datetime, timedelta
+import uuid
+import os
+
+DB_HOST = "10.0.21.51"
+DB_PORT = "27017"
+DB_NAME = "supplychain"
+COLLECTION_NAME = "events"
+
+TARGET_DOCUMENTS = int(os.getenv("GEN_ROWS", "5000"))
+BATCH_SIZE = min(5000, max(500, TARGET_DOCUMENTS))
+
+EVENT_TYPES = ["INSERT", "UPDATE", "DELETE", "CREATE", "MODIFY"]
+REGIONS = ["EU", "APAC", "LATAM", "NA", "EMEA"]
+SOURCES = ["ERP", "WMS", "CRM", "SCM", "TMS"]
+
+FIRST = ["Sophie", "Liam", "Emma", "Noah", "Julia", "Lucas", "Mila", "Daan", "Anna", "Sem",
+         "Eva", "Finn", "Tess", "Bram", "Lotte", "Max", "Sara", "Thijs", "Nina", "Ruben"]
+LAST = ["de Vries", "Jansen", "Bakker", "Visser", "Smit", "Meijer", "Mulder", "Bos",
+        "Vos", "Peters", "Hendriks", "Dijkstra", "Kok", "Willems", "Maas", "Koning"]
+DOMAINS = ["example.com", "mail.nl", "acme-corp.eu", "shopmail.com", "contact.io"]
+
+
+def _name():
+    return f"{random.choice(FIRST)} {random.choice(LAST)}"
+
+
+def generate_fake_event():
+    event_id = str(uuid.uuid4())
+    days_ago = random.randint(0, 365)
+    ts = datetime.now() - timedelta(days=days_ago, hours=random.randint(0, 23),
+                                     minutes=random.randint(0, 59))
+    nm = _name()
+    return {
+        "event_id": event_id,
+        "type": random.choice(EVENT_TYPES),
+        "region": random.choice(REGIONS),
+        "source": random.choice(SOURCES),
+        "amount": random.uniform(100.0, 50000.0),
+        "ts": ts,
+        "contact_name": nm,
+        "contact_email": f"{nm.lower().replace(' ', '.')}{random.randint(1, 999)}@{random.choice(DOMAINS)}",
+        "contact_phone": f"+31 6 {random.randint(10, 99)} {random.randint(100000, 999999)}",
+        "payload": "X" * 500,
+    }
+
+
+def main():
+    print(f"Connecting to MongoDB at {DB_HOST}:{DB_PORT}...")
+    client = pymongo.MongoClient(f"mongodb://{DB_HOST}:{DB_PORT}/")
+    collection = client[DB_NAME][COLLECTION_NAME]
+    print(f"Generating {TARGET_DOCUMENTS} events (with PII)... batch={BATCH_SIZE}")
+    total_generated = 0
+    batch = []
+    for i in range(TARGET_DOCUMENTS):
+        batch.append(generate_fake_event())
+        if len(batch) >= BATCH_SIZE:
+            collection.insert_many(batch)
+            total_generated += len(batch)
+            batch = []
+    if batch:
+        collection.insert_many(batch)
+        total_generated += len(batch)
+    client.close()
+    print(f"Completed! Generated {total_generated} events.")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/deploy/airflow/scripts/generate_mysql_employee_data.py b/deploy/airflow/scripts/generate_mysql_employee_data.py
new file mode 100644
index 0000000..596be74
--- /dev/null
+++ b/deploy/airflow/scripts/generate_mysql_employee_data.py
@@ -0,0 +1,103 @@
+#!/usr/bin/env python3
+"""
+Generate fake employee event data for MySQL.
+Light + configurable via GEN_ROWS. Now also populates realistic PII columns
+(employee_name/email/phone, national_id, home_address, date_of_birth).
+"""
+
+import mysql.connector
+import random
+from datetime import datetime, timedelta
+import uuid
+import os
+
+DB_HOST = "10.0.21.51"
+DB_PORT = "3306"
+DB_NAME = "hr"
+DB_USER = "mo"
+DB_PASSWORD = "Dell2026!"
+
+TARGET_ROWS = int(os.getenv("GEN_ROWS", "5000"))
+BATCH_SIZE = min(10000, max(500, TARGET_ROWS))
+
+DEPARTMENTS = ["HR", "Operations", "Sales", "Marketing", "Finance", "IT", "Engineering", "Legal"]
+ROLE_NAMES = ["Analyst", "Lead", "Manager", "Consultant", "Director", "Engineer", "Specialist", "Coordinator"]
+REGIONS = ["EU", "APAC", "LATAM", "NA", "EMEA"]
+EVENT_TYPES = ["TRANSFER", "PROMOTION", "TERMINATION", "HIRED", "SALARY_CHANGE", "DEPARTMENT_CHANGE"]
+
+FIRST = ["Sophie", "Liam", "Emma", "Noah", "Julia", "Lucas", "Mila", "Daan", "Anna", "Sem",
+         "Eva", "Finn", "Tess", "Bram", "Lotte", "Max", "Sara", "Thijs", "Nina", "Ruben"]
+LAST = ["de Vries", "Jansen", "Bakker", "Visser", "Smit", "Meijer", "Mulder", "Bos",
+        "Vos", "Peters", "Hendriks", "Dijkstra", "Kok", "Willems", "Maas", "Koning"]
+STREETS = ["Kerkstraat", "Dorpsweg", "Molenpad", "Stationsplein", "Industrieweg", "Lindelaan",
+           "Beukenhof", "Havenstraat", "Parkweg", "Schoolstraat"]
+CITIES = ["Amsterdam", "Rotterdam", "Utrecht", "Eindhoven", "Den Haag", "Groningen", "Tilburg", "Almere"]
+DOMAINS = ["example.com", "mail.nl", "acme-corp.eu", "company.eu", "contact.io"]
+
+
+def _name():
+    return f"{random.choice(FIRST)} {random.choice(LAST)}"
+
+
+def _pii():
+    nm = _name()
+    return (
+        nm,
+        f"{nm.lower().replace(' ', '.')}{random.randint(1, 999)}@{random.choice(DOMAINS)}",
+        f"+31 6 {random.randint(10, 99)} {random.randint(100000, 999999)}",
+        str(random.randint(100000000, 999999999)),
+        f"{random.choice(STREETS)} {random.randint(1, 320)}, {random.randint(1000, 9999)} {random.choice(CITIES)}",
+        f"{random.randint(1960, 2002)}-{random.randint(1, 12):02d}-{random.randint(1, 28):02d}",
+    )
+
+
+def generate_fake_employee_event():
+    employee_id = random.randint(1, 100000)
+    department = random.choice(DEPARTMENTS)
+    role_name = random.choice(ROLE_NAMES)
+    region = random.choice(REGIONS)
+    event_type = random.choice(EVENT_TYPES)
+    days_ago = random.randint(0, 730)
+    event_ts = datetime.now() - timedelta(days=days_ago, hours=random.randint(0, 23),
+                                          minutes=random.randint(0, 59))
+    salary_change = round(random.uniform(1000.0, 20000.0), 2) if random.random() > 0.3 else None
+    notes = str(uuid.uuid4()) * 10
+    name, email, phone, nid, address, dob = _pii()
+    return (employee_id, department, role_name, region, event_type, salary_change, event_ts, notes,
+            name, email, phone, nid, address, dob)
+
+
+INSERT_SQL = """
+    INSERT INTO employee_events (employee_id, department, role_name, region,
+                                event_type, salary_change, event_ts, notes,
+                                employee_name, employee_email, employee_phone,
+                                national_id, home_address, date_of_birth)
+    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+"""
+
+
+def main():
+    print(f"Connecting to MySQL at {DB_HOST}:{DB_PORT}...")
+    conn = mysql.connector.connect(host=DB_HOST, port=DB_PORT, database=DB_NAME, user=DB_USER, password=DB_PASSWORD)
+    cursor = conn.cursor()
+    print(f"Generating {TARGET_ROWS} employee events (with PII)... batch={BATCH_SIZE}")
+    total_generated = 0
+    batch = []
+    for i in range(TARGET_ROWS):
+        batch.append(generate_fake_employee_event())
+        if len(batch) >= BATCH_SIZE:
+            cursor.executemany(INSERT_SQL, batch)
+            conn.commit()
+            total_generated += len(batch)
+            batch = []
+    if batch:
+        cursor.executemany(INSERT_SQL, batch)
+        conn.commit()
+        total_generated += len(batch)
+    cursor.close()
+    conn.close()
+    print(f"Completed! Generated {total_generated} employee events.")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/deploy/airflow/scripts/generate_postgres_sales_data.py b/deploy/airflow/scripts/generate_postgres_sales_data.py
new file mode 100644
index 0000000..96acca2
--- /dev/null
+++ b/deploy/airflow/scripts/generate_postgres_sales_data.py
@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+"""
+Generate fake sales order data for PostgreSQL.
+Light + configurable via GEN_ROWS. Now also populates realistic PII columns
+(customer_name/email/phone/ip, billing_iban, shipping_address) so the masking
+ETL + OpenMetadata PII auto-classification have real sensitive data to work on.
+"""
+
+import psycopg2
+import random
+from datetime import datetime, timedelta
+import uuid
+import os
+
+DB_HOST = "10.0.21.51"
+DB_PORT = "5432"
+DB_NAME = "postgres"
+DB_USER = "mo"
+DB_PASSWORD = "Dell2026!"
+
+TARGET_ROWS = int(os.getenv("GEN_ROWS", "5000"))
+BATCH_SIZE = min(10000, max(500, TARGET_ROWS))
+
+REGIONS = ["EU", "APAC", "LATAM", "NA", "EMEA"]
+SALES_CHANNELS = ["STORE", "ONLINE", "MOBILE", "B2B"]
+CURRENCIES = ["EUR", "USD", "GBP", "JPY", "CNY"]
+ORDER_STATUSES = ["SHIPPED", "PENDING", "CANCELLED", "RETURNED", "DELIVERED"]
+
+# ── lightweight PII generator (no external Faker dependency) ─────────────────
+FIRST = ["Sophie", "Liam", "Emma", "Noah", "Julia", "Lucas", "Mila", "Daan", "Anna", "Sem",
+         "Eva", "Finn", "Tess", "Bram", "Lotte", "Max", "Sara", "Thijs", "Nina", "Ruben"]
+LAST = ["de Vries", "Jansen", "Bakker", "Visser", "Smit", "Meijer", "Mulder", "Bos",
+        "Vos", "Peters", "Hendriks", "Dijkstra", "Kok", "Willems", "Maas", "Koning"]
+STREETS = ["Kerkstraat", "Dorpsweg", "Molenpad", "Stationsplein", "Industrieweg", "Lindelaan",
+           "Beukenhof", "Havenstraat", "Parkweg", "Schoolstraat"]
+CITIES = ["Amsterdam", "Rotterdam", "Utrecht", "Eindhoven", "Den Haag", "Groningen", "Tilburg", "Almere"]
+DOMAINS = ["example.com", "mail.nl", "acme-corp.eu", "shopmail.com", "contact.io"]
+
+
+def _name():
+    return f"{random.choice(FIRST)} {random.choice(LAST)}"
+
+
+def _pii():
+    nm = _name()
+    return (
+        nm,
+        f"{nm.lower().replace(' ', '.')}{random.randint(1, 999)}@{random.choice(DOMAINS)}",
+        f"+31 6 {random.randint(10, 99)} {random.randint(100000, 999999)}",
+        ".".join(str(random.randint(1, 254)) for _ in range(4)),
+        f"NL{random.randint(10, 99)}DELL{random.randint(10**9, 10**10 - 1)}",
+        f"{random.choice(STREETS)} {random.randint(1, 320)}, {random.randint(1000, 9999)} {random.choice(CITIES)}",
+    )
+
+
+def generate_fake_order():
+    customer_id = random.randint(1, 100000)
+    product_id = random.randint(1, 5000)
+    region = random.choice(REGIONS)
+    sales_channel = random.choice(SALES_CHANNELS)
+    days_ago = random.randint(0, 730)
+    order_ts = datetime.now() - timedelta(days=days_ago, hours=random.randint(0, 23),
+                                          minutes=random.randint(0, 59))
+    amount = round(random.uniform(10.0, 10000.0), 2)
+    currency = random.choice(CURRENCIES)
+    order_status = random.choice(ORDER_STATUSES)
+    notes = str(uuid.uuid4()) * 10
+    name, email, phone, ip, iban, address = _pii()
+    return (customer_id, product_id, region, sales_channel, order_ts,
+            amount, currency, order_status, notes,
+            name, email, phone, ip, iban, address)
+
+
+INSERT_SQL = """
+    INSERT INTO sales_orders (customer_id, product_id, region, sales_channel,
+                             order_ts, amount, currency, order_status, notes,
+                             customer_name, customer_email, customer_phone,
+                             customer_ip, billing_iban, shipping_address)
+    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+"""
+
+
+def main():
+    print(f"Connecting to PostgreSQL at {DB_HOST}:{DB_PORT}...")
+    conn = psycopg2.connect(host=DB_HOST, port=DB_PORT, database=DB_NAME, user=DB_USER, password=DB_PASSWORD)
+    cursor = conn.cursor()
+    print(f"Generating {TARGET_ROWS} sales orders (with PII)... batch={BATCH_SIZE}")
+    total_generated = 0
+    batch = []
+    for i in range(TARGET_ROWS):
+        batch.append(generate_fake_order())
+        if len(batch) >= BATCH_SIZE:
+            cursor.executemany(INSERT_SQL, batch)
+            conn.commit()
+            total_generated += len(batch)
+            batch = []
+    if batch:
+        cursor.executemany(INSERT_SQL, batch)
+        conn.commit()
+        total_generated += len(batch)
+    cursor.close()
+    conn.close()
+    print(f"Completed! Generated {total_generated} sales orders.")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/deploy/mask_pii_setup.py b/deploy/mask_pii_setup.py
new file mode 100644
index 0000000..9cb0b6e
--- /dev/null
+++ b/deploy/mask_pii_setup.py
@@ -0,0 +1,138 @@
+#!/usr/bin/env python3
+"""Add realistic PII columns to the source tables and backfill a bounded sample.
+
+Runs inside the atc-agents api container (has psycopg2/pymysql/pymongo + network
+to 10.0.21.51). Idempotent: ADD COLUMN IF NOT EXISTS / duplicate-column tolerant.
+Backfills the most recent N rows so OpenMetadata auto-classification samples real
+PII values. New generator rows also get PII (see updated generator scripts).
+"""
+import os
+import random
+
+import psycopg2
+import pymysql
+import pymongo
+
+DB_HOST = os.getenv("SRC_DB_HOST", "10.0.21.51")
+DB_USER = os.getenv("SRC_DB_USER", "mo")
+DB_PASS = os.getenv("SRC_DB_PASSWORD", "Dell2026!")
+N = int(os.getenv("PII_BACKFILL_ROWS", "4000"))
+
+FIRST = ["Sophie", "Liam", "Emma", "Noah", "Julia", "Lucas", "Mila", "Daan", "Anna", "Sem",
+         "Eva", "Finn", "Tess", "Bram", "Lotte", "Max", "Sara", "Thijs", "Nina", "Ruben"]
+LAST = ["de Vries", "Jansen", "Bakker", "Visser", "Smit", "Meijer", "Mulder", "Bos",
+        "Vos", "Peters", "Hendriks", "Dijkstra", "Kok", "Willems", "Maas", "Koning"]
+STREETS = ["Kerkstraat", "Dorpsweg", "Molenpad", "Stationsplein", "Industrieweg", "Lindelaan",
+           "Beukenhof", "Havenstraat", "Parkweg", "Schoolstraat"]
+CITIES = ["Amsterdam", "Rotterdam", "Utrecht", "Eindhoven", "Den Haag", "Groningen", "Tilburg", "Almere"]
+DOMAINS = ["example.com", "mail.nl", "acme-corp.eu", "shopmail.com", "contact.io"]
+
+
+def full_name():
+    return f"{random.choice(FIRST)} {random.choice(LAST)}"
+
+
+def email(name):
+    base = name.lower().replace(" ", ".")
+    return f"{base}{random.randint(1, 999)}@{random.choice(DOMAINS)}"
+
+
+def phone():
+    return f"+31 6 {random.randint(10, 99)} {random.randint(100000, 999999)}"
+
+
+def ip():
+    return ".".join(str(random.randint(1, 254)) for _ in range(4))
+
+
+def iban():
+    return f"NL{random.randint(10, 99)}DELL{random.randint(10**9, 10**10 - 1)}"
+
+
+def national_id():
+    return str(random.randint(100000000, 999999999))
+
+
+def address():
+    return f"{random.choice(STREETS)} {random.randint(1, 320)}, {random.randint(1000, 9999)} {random.choice(CITIES)}"
+
+
+def dob():
+    return f"{random.randint(1960, 2002)}-{random.randint(1, 12):02d}-{random.randint(1, 28):02d}"
+
+
+def do_postgres():
+    print("== PostgreSQL sales_orders ==")
+    conn = psycopg2.connect(host=DB_HOST, dbname="postgres", user=DB_USER, password=DB_PASS, port=5432, connect_timeout=10)
+    conn.autocommit = True
+    cur = conn.cursor()
+    for col, typ in [("customer_name", "text"), ("customer_email", "text"), ("customer_phone", "text"),
+                     ("customer_ip", "text"), ("billing_iban", "text"), ("shipping_address", "text")]:
+        cur.execute(f"ALTER TABLE public.sales_orders ADD COLUMN IF NOT EXISTS {col} {typ}")
+    print("  columns ensured")
+    cur.execute("SELECT order_id FROM public.sales_orders ORDER BY order_id DESC LIMIT %s", (N,))
+    ids = [r[0] for r in cur.fetchall()]
+    rows = []
+    for oid in ids:
+        nm = full_name()
+        rows.append((nm, email(nm), phone(), ip(), iban(), address(), oid))
+    cur.executemany(
+        "UPDATE public.sales_orders SET customer_name=%s, customer_email=%s, customer_phone=%s, "
+        "customer_ip=%s, billing_iban=%s, shipping_address=%s WHERE order_id=%s", rows)
+    print(f"  backfilled {len(rows)} rows with PII")
+    cur.close()
+    conn.close()
+
+
+def do_mysql():
+    print("== MySQL employee_events ==")
+    conn = pymysql.connect(host=DB_HOST, user=DB_USER, password=DB_PASS, database="hr", port=3306,
+                           connect_timeout=10, autocommit=True)
+    cur = conn.cursor()
+    for col, typ in [("employee_name", "varchar(120)"), ("employee_email", "varchar(160)"),
+                     ("employee_phone", "varchar(40)"), ("national_id", "varchar(20)"),
+                     ("home_address", "varchar(255)"), ("date_of_birth", "date")]:
+        try:
+            cur.execute(f"ALTER TABLE employee_events ADD COLUMN {col} {typ}")
+        except Exception as e:
+            if "Duplicate column" not in str(e):
+                raise
+    print("  columns ensured")
+    cur.execute("SELECT event_id FROM employee_events ORDER BY event_id DESC LIMIT %s", (N,))
+    ids = [r[0] for r in cur.fetchall()]
+    rows = []
+    for eid in ids:
+        nm = full_name()
+        rows.append((nm, email(nm), phone(), national_id(), address(), dob(), eid))
+    cur.executemany(
+        "UPDATE employee_events SET employee_name=%s, employee_email=%s, employee_phone=%s, "
+        "national_id=%s, home_address=%s, date_of_birth=%s WHERE event_id=%s", rows)
+    print(f"  backfilled {len(rows)} rows with PII")
+    cur.close()
+    conn.close()
+
+
+def do_mongo():
+    print("== MongoDB events ==")
+    uri = os.getenv("SRC_MONGO_URI", f"mongodb://{DB_HOST}:27017/?replicaSet=rs0")
+    client = pymongo.MongoClient(uri, serverSelectionTimeoutMS=10000)
+    coll = client["supplychain"]["events"]
+    docs = list(coll.find({}, {"_id": 1}).sort("_id", -1).limit(N))
+    n = 0
+    for d in docs:
+        nm = full_name()
+        coll.update_one({"_id": d["_id"]}, {"$set": {
+            "contact_name": nm, "contact_email": email(nm), "contact_phone": phone(),
+        }})
+        n += 1
+    print(f"  updated {n} docs with PII fields")
+    client.close()
+
+
+if __name__ == "__main__":
+    for fn in (do_postgres, do_mysql, do_mongo):
+        try:
+            fn()
+        except Exception as e:
+            print(f"  ERROR in {fn.__name__}: {e}")
+    print("done")