SCD Type 2: реализация через MERGE INTO и Iceberg row-level deletes

CREATE TABLE dim_customers (
    -- Бизнес-ключ (натуральный ключ из источника):
    customer_id     BIGINT        NOT NULL,

    -- Бизнес-атрибуты (то, что изменяется):
    email           STRING,
    full_name       STRING,
    country         STRING,
    city            STRING,
    loyalty_tier    STRING,        -- bronze / silver / gold

    -- SCD2 технические поля:
    valid_from      TIMESTAMP     NOT NULL,  -- начало действия версии
    valid_to        TIMESTAMP,               -- конец действия, NULL = текущая
    is_current      BOOLEAN       NOT NULL,  -- быстрый фильтр текущих записей
    surrogate_key   BIGINT        NOT NULL,  -- суррогатный PK для JOIN с фактами
    record_hash     STRING        NOT NULL,  -- SHA-256 атрибутов (для change detection)

    -- Аудит:
    created_at      TIMESTAMP     NOT NULL,  -- когда строка появилась в DWH
    source_system   STRING,                  -- откуда пришли данные
    batch_id        STRING                   -- ID ETL-запуска, создавшего строку
)
USING iceberg
PARTITIONED BY (is_current, days(valid_from))
TBLPROPERTIES (
    'write.format.default' = 'parquet',
    'write.parquet.compression-codec' = 'zstd',
    'format-version' = '2'
);

# Включить MoR для таблицы измерений с частыми обновлениями:
spark.sql("""
    ALTER TABLE catalog.dwh.dim_customers SET TBLPROPERTIES (
        'write.delete.mode'          = 'merge-on-read',
        'write.update.mode'          = 'merge-on-read',
        'write.merge.mode'           = 'merge-on-read',
        -- Автоматическая компакция после N операций:
        'write.target-file-size-bytes' = '134217728',   -- 128 MB target
        'write.distribution-mode'    = 'hash'           -- hash by partition key
    )
""")

from pyspark.sql import functions as F
from pyspark.sql import SparkSession, DataFrame
from typing import List


def add_record_hash(df: DataFrame, columns: List[str]) -> DataFrame:
    """
    Вычислить SHA-256 хеш от конкатенации указанных колонок.

    Важные детали:
    - COALESCE(col, '') для NULL-safety: sha2('a|NULL|b') != sha2('a||b')
      при разных NULL-подходах, поэтому нормализуем NULL → пустая строка.
    - Разделитель '|' выбран так, чтобы не встречаться в данных.
      Для надёжности можно использовать chr(1) как non-printable separator.
    - sha2(..., 256) возвращает 64-символьный hex string.
    - xxhash64 быстрее, но может иметь коллизии на больших объёмах.
    """
    normalized = [F.coalesce(F.col(c).cast("string"), F.lit("")).alias(c)
                  for c in columns]
    concat_expr = F.concat_ws("|", *[F.col(c) for c in columns])

    return (
        df
        .withColumn("_norm_concat", concat_expr)
        .withColumn("record_hash", F.sha2(F.col("_norm_concat"), 256))
        .drop("_norm_concat")
    )


# Пример использования:
TRACKED_COLUMNS = ["email", "full_name", "country", "city", "loyalty_tier"]

source_with_hash = add_record_hash(source_df, TRACKED_COLUMNS)

from pyspark.sql import functions as F, SparkSession, DataFrame
from pyspark.sql.types import BooleanType, TimestampType, StringType, LongType
from datetime import datetime, timezone
import xxhash


TRACKED_COLUMNS = ["email", "full_name", "country", "city", "loyalty_tier"]


def build_source_expanded(
    spark: SparkSession,
    source_df: DataFrame,
    target_table: str,
    now: str = "current_timestamp()"
) -> DataFrame:
    """
    Source Expansion для SCD Type 2:
    - Новые записи   → одна строка с action='insert'
    - Изменившиеся  → две строки: action='close' + action='insert'
    - Без изменений → пропускаются
    """

    source_with_hash = add_record_hash(source_df, TRACKED_COLUMNS)
    source_with_hash.createOrReplaceTempView("_scd2_source")

    # Найти текущие версии из target:
    current_target = spark.sql(f"""
        SELECT customer_id, surrogate_key, record_hash
        FROM {target_table}
        WHERE is_current = true
    """)
    current_target.createOrReplaceTempView("_scd2_current")

    # Классифицировать изменения:
    classified = spark.sql(f"""
        SELECT
            s.*,
            CASE
                WHEN c.customer_id IS NULL         THEN 'new'
                WHEN s.record_hash != c.record_hash THEN 'changed'
                ELSE 'unchanged'
            END AS _change_type,
            c.surrogate_key AS _old_surrogate_key
        FROM _scd2_source s
        LEFT JOIN _scd2_current c ON s.customer_id = c.customer_id
    """)

    # Строки для закрытия старых версий (только changed):
    rows_to_close = classified.filter(
        F.col("_change_type") == "changed"
    ).select(
        F.col("_old_surrogate_key").alias("surrogate_key"),
        F.col("customer_id"),
        F.lit("close").alias("_action"),
        F.lit(None).cast("string").alias("email"),
        F.lit(None).cast("string").alias("full_name"),
        F.lit(None).cast("string").alias("country"),
        F.lit(None).cast("string").alias("city"),
        F.lit(None).cast("string").alias("loyalty_tier"),
        F.lit(None).cast("string").alias("record_hash"),
    )

    # Строки для вставки новых версий (new + changed):
    rows_to_insert = classified.filter(
        F.col("_change_type").isin("new", "changed")
    ).select(
        # Детерминированный surrogate_key через хеш business_key + valid_from:
        F.xxhash64(
            F.col("customer_id"),
            F.expr(now)
        ).alias("surrogate_key"),
        F.col("customer_id"),
        F.lit("insert").alias("_action"),
        F.col("email"),
        F.col("full_name"),
        F.col("country"),
        F.col("city"),
        F.col("loyalty_tier"),
        F.col("record_hash"),
    )

    return rows_to_close.union(rows_to_insert)

from pyspark.sql import SparkSession, DataFrame
from pyspark.sql import functions as F
from dataclasses import dataclass
from typing import List, Optional
import logging

logger = logging.getLogger(__name__)


@dataclass
class SCD2Config:
    """Конфигурация SCD Type 2 пайплайна."""
    target_table: str            # catalog.schema.table_name
    business_key: str            # колонка натурального ключа
    tracked_columns: List[str]   # атрибуты, изменение которых создаёт новую версию
    audit_columns: List[str]     # колонки, которые не триггерят новую версию
    batch_id: str                # ID текущего batch для аудита
    source_system: str = "crm"


def run_scd2_pipeline(
    spark: SparkSession,
    source_df: DataFrame,
    config: SCD2Config
) -> dict:
    """
    Двухшаговый SCD Type 2:
    1. Закрыть устаревшие версии (MERGE / UPDATE)
    2. Вставить новые версии (INSERT)

    Возвращает метрики: rows_closed, rows_inserted, rows_unchanged.
    """

    # Шаг 0: Вычислить хеши и подготовить source
    source_hashed = add_record_hash(source_df, config.tracked_columns)

    # Зафиксировать timestamp для всего batch (консистентность):
    # Используем одно значение NOW для всех строк в batch
    batch_ts = spark.sql("SELECT current_timestamp() AS ts").first()["ts"]
    batch_ts_str = f"TIMESTAMP '{batch_ts}'"

    source_hashed = source_hashed.withColumn("_batch_ts", F.lit(batch_ts))
    source_hashed.createOrReplaceTempView("_scd2_source")

    # -----------------------------------------------------------------------
    # Шаг 1: Закрыть устаревшие версии
    # Найти записи в target, для которых в source пришёл изменённый hash
    # -----------------------------------------------------------------------
    logger.info("SCD2 Step 1: Closing outdated records")

    close_sql = f"""
        MERGE INTO {config.target_table} AS target
        USING (
            SELECT
                s.{config.business_key},
                s.record_hash AS new_hash,
                s._batch_ts   AS close_ts
            FROM _scd2_source s
        ) AS src
        ON  target.{config.business_key} = src.{config.business_key}
        AND target.is_current = true
        AND target.record_hash != src.new_hash   -- реально изменилось
        WHEN MATCHED THEN UPDATE SET
            target.is_current = false,
            target.valid_to   = src.close_ts
    """

    spark.sql(close_sql)

    closed_count = spark.sql(f"""
        SELECT COUNT(*)
        FROM {config.target_table}
        WHERE is_current = false
          AND valid_to = {batch_ts_str}
    """).first()[0]

    logger.info(f"SCD2 Step 1: Closed {closed_count} records")

    # -----------------------------------------------------------------------
    # Шаг 2: Вставить новые версии
    # new = не было в target | changed = только что закрыли на шаге 1
    # -----------------------------------------------------------------------
    logger.info("SCD2 Step 2: Inserting new versions")

    insert_sql = f"""
        INSERT INTO {config.target_table}
        SELECT
            s.{config.business_key},
            s.email,
            s.full_name,
            s.country,
            s.city,
            s.loyalty_tier,
            s._batch_ts                                    AS valid_from,
            CAST(NULL AS TIMESTAMP)                        AS valid_to,
            true                                           AS is_current,
            xxhash64(s.{config.business_key}, s._batch_ts) AS surrogate_key,
            s.record_hash,
            s._batch_ts                                    AS created_at,
            '{config.source_system}'                       AS source_system,
            '{config.batch_id}'                            AS batch_id
        FROM _scd2_source s
        LEFT JOIN {config.target_table} t
            ON  s.{config.business_key} = t.{config.business_key}
            AND t.is_current = true
        WHERE t.{config.business_key} IS NULL   -- новый customer (нет текущей версии)
           OR t.record_hash != s.record_hash    -- изменился (шаг 1 уже закрыл)
    """

    spark.sql(insert_sql)

    inserted_count = spark.sql(f"""
        SELECT COUNT(*)
        FROM {config.target_table}
        WHERE batch_id = '{config.batch_id}'
          AND is_current = true
    """).first()[0]

    unchanged_count = source_df.count() - closed_count

    logger.info(f"SCD2 Step 2: Inserted {inserted_count} new versions")

    return {
        "closed":    closed_count,
        "inserted":  inserted_count,
        "unchanged": unchanged_count,
        "batch_ts":  str(batch_ts),
    }

# ❌ Антипаттерн 1: monotonically_increasing_id()
# Не детерминирован при retry, зависит от порядка партиций
df.withColumn("surrogate_key", F.monotonically_increasing_id())

# ❌ Антипаттерн 2: UUID
# UUID v4 случаен - разные значения при retry, нет idempotency
import uuid
df.withColumn("surrogate_key", F.lit(str(uuid.uuid4())))

# ✅ Подход 1: xxhash64 от business_key + valid_from
# Детерминирован, быстр (non-cryptographic hash)
df.withColumn(
    "surrogate_key",
    F.xxhash64(F.col("customer_id"), F.col("valid_from"))
)

# ✅ Подход 2: hash64 через sha2 + бит-усечение (для совместимости с BIGINT)
# SHA-256 → берём первые 16 hex символов → приводим к int64
df.withColumn(
    "surrogate_key",
    F.conv(F.substring(F.sha2(
        F.concat_ws("|", F.col("customer_id").cast("string"),
                    F.col("valid_from").cast("string")), 256
    ), 1, 15), 16, 10).cast("long")
)

# ✅ Подход 3: Iceberg sequence number (автоинкремент на уровне snapshot)
# Доступен через метаданные Iceberg, уникален в рамках таблицы
# Не подходит для cross-table joins без дополнительного маппинга

from pyspark.sql import functions as F
from pyspark.sql.window import Window


def deduplicate_within_batch(
    df: DataFrame,
    business_key: str,
    event_ts_col: str,
    tracked_cols: List[str]
) -> DataFrame:
    """
    Если в batch несколько событий для одного business_key:
    - Взять последнее состояние для финального INSERT в target
    - Сохранить промежуточные версии если event_ts известен

    Для простоты: берём последнее событие (если промежуточные версии не нужны).
    """
    window = Window.partitionBy(business_key).orderBy(F.col(event_ts_col).desc())

    return (
        df
        .withColumn("_row_num", F.row_number().over(window))
        .filter(F.col("_row_num") == 1)
        .drop("_row_num")
    )

def handle_late_arriving(
    spark: SparkSession,
    source_df: DataFrame,
    target_table: str,
    event_ts_col: str
) -> None:
    """
    Late-arriving SCD Type 2:
    Использует event_timestamp как valid_from вместо processing_time.

    Проблема: если запись уже была закрыта позже event_ts,
    нужно «вклинить» новую версию в середину истории.
    """
    source_df.createOrReplaceTempView("_late_source")

    # Найти пересекающиеся интервалы в target:
    overlaps = spark.sql(f"""
        SELECT
            t.customer_id,
            t.surrogate_key,
            t.valid_from AS t_valid_from,
            t.valid_to   AS t_valid_to,
            s.{event_ts_col} AS event_ts
        FROM {target_table} t
        JOIN _late_source s
            ON t.customer_id = s.customer_id
           AND s.{event_ts_col} BETWEEN t.valid_from
               AND COALESCE(t.valid_to, TIMESTAMP '9999-01-01')
        WHERE t.is_current = false OR t.valid_to IS NULL
    """)

    if overlaps.count() > 0:
        logger.warning(
            f"Late-arriving data: {overlaps.count()} records "
            f"require timeline surgery - inserting mid-history versions"
        )
        # Логика «timeline surgery» - сложна и error-prone.
        # Рекомендация для production: отправлять в DLQ для ручного review.

def check_scd2_integrity(spark: SparkSession, target_table: str) -> dict:
    """
    Проверки целостности SCD Type 2 таблицы:
    1. Ровно одна текущая версия на customer_id
    2. Непрерывные интервалы (no gaps, no overlaps)
    3. is_current соответствует valid_to IS NULL
    """

    # Проверка 1: дублирующиеся текущие версии
    dups = spark.sql(f"""
        SELECT customer_id, COUNT(*) AS cnt
        FROM {target_table}
        WHERE is_current = true
        GROUP BY customer_id
        HAVING COUNT(*) > 1
    """)
    dup_count = dups.count()

    # Проверка 2: несоответствие is_current и valid_to
    inconsistent = spark.sql(f"""
        SELECT COUNT(*) AS cnt
        FROM {target_table}
        WHERE (is_current = true  AND valid_to IS NOT NULL)
           OR (is_current = false AND valid_to IS NULL)
    """).first()[0]

    # Проверка 3: пересекающиеся интервалы
    # Используем self-join (expensive! запускать только в режиме аудита)
    overlaps = spark.sql(f"""
        SELECT COUNT(*) AS cnt
        FROM {target_table} a
        JOIN {target_table} b
            ON  a.customer_id = b.customer_id
            AND a.surrogate_key < b.surrogate_key
            AND a.valid_from < COALESCE(b.valid_to, TIMESTAMP '9999-01-01')
            AND b.valid_from < COALESCE(a.valid_to, TIMESTAMP '9999-01-01')
    """).first()[0]

    integrity_ok = (dup_count == 0 and inconsistent == 0 and overlaps == 0)

    return {
        "integrity_ok":    integrity_ok,
        "duplicate_currents": dup_count,
        "inconsistent_flags": inconsistent,
        "overlapping_intervals": overlaps,
    }

-- Temporal join: факты с атрибутами на момент события
SELECT
    o.order_id,
    o.order_date,
    o.amount,
    c.email,
    c.country,
    c.loyalty_tier     -- tier клиента НА МОМЕНТ заказа
FROM fact_orders o
JOIN dim_customers c
    ON  o.customer_id = c.customer_id
    AND o.order_date >= c.valid_from
    AND (o.order_date <  c.valid_to OR c.valid_to IS NULL);

-- Эквивалент через surrogate_key (быстрее если fact хранит surrogate_key):
SELECT
    o.order_id,
    o.order_date,
    o.amount,
    c.email,
    c.country
FROM fact_orders_with_sk o       -- уже содержит surrogate_key
JOIN dim_customers c
    ON o.customer_sk = c.surrogate_key;

-- Состояние dimension на конкретную дату (point-in-time query):
SELECT *
FROM dim_customers
WHERE customer_id = 42
  AND valid_from <= TIMESTAMP '2024-06-01 00:00:00'
  AND (valid_to > TIMESTAMP '2024-06-01 00:00:00' OR valid_to IS NULL);

-- История изменений конкретного клиента:
SELECT
    customer_id,
    email,
    country,
    loyalty_tier,
    valid_from,
    valid_to,
    DATEDIFF(
        COALESCE(valid_to, current_timestamp()),
        valid_from
    ) AS days_active
FROM dim_customers
WHERE customer_id = 42
ORDER BY valid_from;

-- Текущее состояние (используем partition pruning):
SELECT *
FROM dim_customers
WHERE is_current = true;    -- partition pruning → читается только partition is_current=true

# Оптимизация 1: Broadcast-join для небольшой dimension
from pyspark.sql import functions as F

result = (
    fact_orders
    .join(
        F.broadcast(dim_customers.filter("is_current = true")),
        on="customer_id",
        how="left"
    )
    # Для факта нужны именно текущие атрибуты → broadcast текущего среза
)

# Оптимизация 2: Partition dimension по date ranges для filter pushdown
# Если dim_customers партиционирована по year(valid_from):
spark.sql("""
    SELECT o.*, c.email, c.country
    FROM fact_orders o
    JOIN dim_customers c
        ON  o.customer_id = c.customer_id
        AND o.order_date >= c.valid_from
        AND (o.order_date < c.valid_to OR c.valid_to IS NULL)
        AND YEAR(c.valid_from) BETWEEN 2023 AND 2024  -- partition pruning
""")

# Оптимизация 3: Surrogate key lookup (самый быстрый join)
# Требует pre-computation surrogate_key в fact pipeline
spark.sql("""
    SELECT o.*, c.email, c.country
    FROM fact_orders_enriched o    -- surrogate_key уже вычислен в fact pipeline
    JOIN dim_customers c ON o.customer_dim_sk = c.surrogate_key
""")

from pyspark.sql import SparkSession


def run_scd2_maintenance(
    spark: SparkSession,
    target_table: str,
    max_delete_files_ratio: float = 0.1
) -> None:
    """
    Обслуживание SCD2 Iceberg таблицы.
    Запускать по расписанию (например, 1 раз в сутки).
    """

    # Проверить соотношение delete files к data files:
    stats = spark.sql(f"""
        SELECT
            SUM(CASE WHEN content = 0 THEN 1 ELSE 0 END) AS data_files,
            SUM(CASE WHEN content = 1 THEN 1 ELSE 0 END) AS pos_delete_files,
            SUM(CASE WHEN content = 2 THEN 1 ELSE 0 END) AS eq_delete_files
        FROM {target_table}.files
    """).first()

    delete_ratio = (stats["pos_delete_files"] or 0) / max(stats["data_files"] or 1, 1)

    if delete_ratio > max_delete_files_ratio:
        # Компакция: rewrite data files с применением delete files
        spark.sql(f"""
            CALL spark_catalog.system.rewrite_data_files(
                table => '{target_table}',
                strategy => 'binpack',
                options => map(
                    'target-file-size-bytes', '134217728',
                    'max-concurrent-file-group-rewrites', '5'
                )
            )
        """)

    # Удалить orphan files (файлы без ссылок в метаданных):
    spark.sql(f"""
        CALL spark_catalog.system.remove_orphan_files(
            table => '{target_table}',
            older_than => TIMESTAMP '{spark.sql("SELECT current_timestamp() - INTERVAL 3 DAYS").first()[0]}'
        )
    """)

    # Expire старые snapshots (сохранить последние 14 дней):
    spark.sql(f"""
        CALL spark_catalog.system.expire_snapshots(
            table => '{target_table}',
            older_than => TIMESTAMP '{spark.sql("SELECT current_timestamp() - INTERVAL 14 DAYS").first()[0]}',
            retain_last => 10
        )
    """)

    # Обновить статистику для query optimizer:
    spark.sql(f"ANALYZE TABLE {target_table} COMPUTE STATISTICS")

-- Вариант 1: Только is_current (простой, хорошо для текущих запросов)
PARTITIONED BY (is_current)
-- + Текущий срез: partition pruning → только is_current=true partition
-- - Историческая партиция огромна, нет temporal pruning

-- Вариант 2: is_current + год valid_from (баланс)
PARTITIONED BY (is_current, years(valid_from))
-- + Temporal queries могут использовать pruning по году
-- + Текущие данные изолированы в отдельной партиции
-- - Мелкие партиции если мало изменений в старых годах

-- Вариант 3: Только years(valid_from) (для глубокой истории)
PARTITIONED BY (years(valid_from))
-- + Равномерное распределение по времени
-- - Текущие записи разбросаны по всем партициям (нет partition pruning для is_current)

-- Рекомендация для большинства случаев: Вариант 2

# Z-order по customer_id + valid_from для оптимизации temporal joins:
spark.sql(f"""
    CALL spark_catalog.system.rewrite_data_files(
        table => '{target_table}',
        strategy => 'sort',
        sort_order => 'customer_id ASC NULLS LAST, valid_from ASC NULLS LAST',
        options => map('target-file-size-bytes', '134217728')
    )
""")

# ❌ Проблема: повторный запуск создаёт дубли
spark.sql(f"""
    INSERT INTO dim_customers
    SELECT *, current_timestamp() AS valid_from
    FROM source_changes
""")
# При retry → duplicate rows с разными valid_from

# ✅ Решение: проверять перед вставкой
spark.sql(f"""
    INSERT INTO dim_customers
    SELECT s.*, '{batch_ts}' AS valid_from
    FROM source_changes s
    LEFT JOIN dim_customers t
        ON s.customer_id = t.customer_id
       AND t.batch_id = '{batch_id}'    -- уже вставлено в этом batch?
    WHERE t.customer_id IS NULL         -- не вставлено → вставляем
""")

# ❌ Проблема: current_timestamp() вычисляется для каждой строки отдельно
# На многоузловом кластере строки получают разные timestamps!
df.withColumn("valid_from", F.current_timestamp())

# ✅ Решение: зафиксировать одно значение для всего batch
batch_ts = spark.sql("SELECT current_timestamp() AS ts").first()["ts"]
df.withColumn("valid_from", F.lit(batch_ts))

# ❌ Проблема: valid_to(v1) != valid_from(v2)
# Версия 1: valid_from=10:00:00.001, valid_to=10:00:01.005
# Версия 2: valid_from=10:00:01.006, valid_to=NULL
# Gap между версиями: 10:00:01.005 → 10:00:01.006

# ✅ Решение: закрывать точно тем же timestamp, что открывает следующую
# Шаг 1: вычислить batch_ts
# Шаг 2: в UPDATE SET valid_to = batch_ts
# Шаг 3: в INSERT SET valid_from = batch_ts (тот же timestamp!)
# → valid_to(v1) == valid_from(v2), пробел нулевой

# ❌ Проблема: NULL != NULL в SQL → бесконечные изменения
# Если поле email стало NULL, а в target уже было NULL:
# SHA2("a|NULL|b") vs SHA2("a|NULL|b") могут дать разные результаты
# если NULL сериализуется по-разному

# ✅ Решение: всегда COALESCE(col, '') перед хешированием
F.sha2(
    F.concat_ws("|",
        F.coalesce(F.col("email"),       F.lit("")),
        F.coalesce(F.col("country"),     F.lit("")),
        F.coalesce(F.col("loyalty_tier"),F.lit(""))
    ), 256
)

from pyspark.sql import SparkSession, DataFrame
from pyspark.sql import functions as F
from pyspark.sql.window import Window
from dataclasses import dataclass, field
from typing import List, Optional
import logging
import uuid

logger = logging.getLogger(__name__)


@dataclass
class SCD2PipelineConfig:
    """Полная конфигурация SCD2 пайплайна."""
    # Таблицы:
    source_table:  str  = "catalog.raw.crm_customers"
    target_table:  str  = "catalog.dwh.dim_customers"

    # Ключи и атрибуты:
    business_key:  str  = "customer_id"
    tracked_cols: List[str] = field(default_factory=lambda: [
        "email", "full_name", "country", "city", "loyalty_tier", "phone"
    ])

    # Аудит:
    source_system: str  = "crm"
    batch_id:      str  = field(default_factory=lambda: str(uuid.uuid4()))

    # Производительность:
    broadcast_threshold_mb: int = 512
    enable_integrity_check: bool = True
    enable_maintenance: bool = False   # запускать компакцию?
    maintenance_delete_ratio: float = 0.15


def create_dim_customers_table(spark: SparkSession, cfg: SCD2PipelineConfig) -> None:
    """DDL для SCD2 таблицы (idempotent через IF NOT EXISTS)."""
    spark.sql(f"""
        CREATE TABLE IF NOT EXISTS {cfg.target_table} (
            customer_id   BIGINT        NOT NULL,
            email         STRING,
            full_name     STRING,
            country       STRING,
            city          STRING,
            loyalty_tier  STRING,
            phone         STRING,
            valid_from    TIMESTAMP     NOT NULL,
            valid_to      TIMESTAMP,
            is_current    BOOLEAN       NOT NULL,
            surrogate_key BIGINT        NOT NULL,
            record_hash   STRING        NOT NULL,
            created_at    TIMESTAMP     NOT NULL,
            source_system STRING,
            batch_id      STRING
        )
        USING iceberg
        PARTITIONED BY (is_current, years(valid_from))
        TBLPROPERTIES (
            'write.delete.mode'  = 'merge-on-read',
            'write.update.mode'  = 'merge-on-read',
            'write.merge.mode'   = 'merge-on-read',
            'format-version'     = '2',
            'write.parquet.compression-codec' = 'zstd'
        )
    """)
    logger.info(f"Table {cfg.target_table} ensured")


def load_and_prepare_source(
    spark: SparkSession,
    cfg: SCD2PipelineConfig
) -> DataFrame:
    """
    Загрузить источник и подготовить к SCD2:
    1. Прочитать из raw слоя
    2. Дедуплицировать (оставить последнее событие на customer_id)
    3. Вычислить record_hash
    """
    raw = spark.table(cfg.source_table)

    # Дедупликация: оставить последнее событие для каждого customer_id
    w = Window.partitionBy(cfg.business_key).orderBy(
        F.col("event_timestamp").desc(),
        F.col("_file_modification_time").desc()  # tiebreak
    )
    deduped = (
        raw
        .withColumn("_rn", F.row_number().over(w))
        .filter(F.col("_rn") == 1)
        .drop("_rn", "event_timestamp", "_file_modification_time")
    )

    # Вычислить record_hash
    hash_expr = F.sha2(
        F.concat_ws("|", *[
            F.coalesce(F.col(c).cast("string"), F.lit(""))
            for c in cfg.tracked_cols
        ]), 256
    )
    return deduped.withColumn("record_hash", hash_expr)


def step1_close_outdated(
    spark: SparkSession,
    source_df: DataFrame,
    cfg: SCD2PipelineConfig,
    batch_ts: str
) -> int:
    """
    Шаг 1: Закрыть устаревшие версии.
    MERGE INTO: найти текущие записи с изменившимся hash → UPDATE valid_to, is_current=false.
    """
    source_df.createOrReplaceTempView("_scd2_src_step1")

    spark.sql(f"""
        MERGE INTO {cfg.target_table} AS target
        USING (
            SELECT {cfg.business_key}, record_hash
            FROM _scd2_src_step1
        ) AS src
        ON  target.{cfg.business_key} = src.{cfg.business_key}
        AND target.is_current          = true
        AND target.record_hash        != src.record_hash
        WHEN MATCHED THEN UPDATE SET
            target.is_current = false,
            target.valid_to   = TIMESTAMP '{batch_ts}'
    """)

    closed = spark.sql(f"""
        SELECT COUNT(*) FROM {cfg.target_table}
        WHERE is_current = false
          AND valid_to = TIMESTAMP '{batch_ts}'
          AND batch_id != '{cfg.batch_id}'
    """).first()[0]

    logger.info(f"[SCD2] Step 1 CLOSE: {closed} records closed, batch_ts={batch_ts}")
    return closed


def step2_insert_new_versions(
    spark: SparkSession,
    source_df: DataFrame,
    cfg: SCD2PipelineConfig,
    batch_ts: str
) -> int:
    """
    Шаг 2: Вставить новые версии.
    INSERT: новые customer + те, что были закрыты на шаге 1.
    """
    source_df.createOrReplaceTempView("_scd2_src_step2")

    spark.sql(f"""
        INSERT INTO {cfg.target_table}
        SELECT
            s.{cfg.business_key},
            s.email,
            s.full_name,
            s.country,
            s.city,
            s.loyalty_tier,
            s.phone,
            TIMESTAMP '{batch_ts}'                              AS valid_from,
            CAST(NULL AS TIMESTAMP)                            AS valid_to,
            true                                               AS is_current,
            xxhash64(s.{cfg.business_key},
                     CAST('{batch_ts}' AS TIMESTAMP))          AS surrogate_key,
            s.record_hash,
            TIMESTAMP '{batch_ts}'                             AS created_at,
            '{cfg.source_system}'                              AS source_system,
            '{cfg.batch_id}'                                   AS batch_id
        FROM _scd2_src_step2 s
        LEFT JOIN {cfg.target_table} t
            ON  s.{cfg.business_key} = t.{cfg.business_key}
            AND t.is_current = true
        WHERE t.{cfg.business_key} IS NULL        -- новый customer
           OR t.record_hash != s.record_hash      -- изменился (закрыт на шаге 1)
    """)

    inserted = spark.sql(f"""
        SELECT COUNT(*) FROM {cfg.target_table}
        WHERE batch_id = '{cfg.batch_id}'
          AND is_current = true
    """).first()[0]

    logger.info(f"[SCD2] Step 2 INSERT: {inserted} new versions inserted")
    return inserted


def run_scd2_full_pipeline(
    spark: SparkSession,
    cfg: SCD2PipelineConfig
) -> dict:
    """
    Точка входа: полный SCD Type 2 пайплайн.
    """
    logger.info(f"[SCD2] Starting pipeline batch_id={cfg.batch_id}")

    # DDL
    create_dim_customers_table(spark, cfg)

    # Подготовить source
    source_df = load_and_prepare_source(spark, cfg)
    source_count = source_df.count()
    source_df.cache()
    logger.info(f"[SCD2] Source records: {source_count}")

    # Зафиксировать batch_ts для консистентности всего batch
    batch_ts = spark.sql("SELECT current_timestamp() AS ts").first()["ts"]
    batch_ts_str = str(batch_ts)

    # Шаг 1: Закрыть устаревшие
    closed = step1_close_outdated(spark, source_df, cfg, batch_ts_str)

    # Шаг 2: Вставить новые версии
    inserted = step2_insert_new_versions(spark, source_df, cfg, batch_ts_str)

    unchanged = source_count - inserted

    # Проверка целостности
    if cfg.enable_integrity_check:
        integrity = check_scd2_integrity(spark, cfg.target_table)
        if not integrity["integrity_ok"]:
            logger.error(f"[SCD2] Integrity check FAILED: {integrity}")
            raise RuntimeError(f"SCD2 integrity violation: {integrity}")
        logger.info("[SCD2] Integrity check passed")

    # Обслуживание (если включено)
    if cfg.enable_maintenance:
        run_scd2_maintenance(spark, cfg.target_table,
                             cfg.maintenance_delete_ratio)

    source_df.unpersist()

    metrics = {
        "batch_id":   cfg.batch_id,
        "batch_ts":   batch_ts_str,
        "source":     source_count,
        "closed":     closed,
        "inserted":   inserted,
        "unchanged":  unchanged,
    }
    logger.info(f"[SCD2] Pipeline completed: {metrics}")
    return metrics

from dataclasses import dataclass
from typing import Optional


@dataclass
class SCD2Metrics:
    """Метрики одного SCD2 batch для мониторинга."""
    batch_id:         str
    source_count:     int
    closed_count:     int
    inserted_count:   int
    unchanged_count:  int
    batch_duration_s: float

    @property
    def change_rate(self) -> float:
        """Доля изменившихся записей. Аномалии: внезапный рост/падение."""
        return self.inserted_count / max(self.source_count, 1)

    @property
    def insert_to_close_ratio(self) -> float:
        """
        Для нормального SCD2: inserted ≈ closed (каждая закрытая → новая).
        Аномалия: inserted >> closed (дубли?) или inserted << closed (потеря данных?).
        """
        return self.inserted_count / max(self.closed_count, 1)

    def alert_conditions(self) -> list:
        alerts = []
        if self.change_rate > 0.5:  # >50% записей изменились за batch
            alerts.append(f"HIGH_CHANGE_RATE: {self.change_rate:.1%}")
        if self.insert_to_close_ratio > 1.2 or self.insert_to_close_ratio < 0.8:
            alerts.append(f"INSERT_CLOSE_MISMATCH: ratio={self.insert_to_close_ratio:.2f}")
        if self.source_count == 0:
            alerts.append("EMPTY_SOURCE: нет входных данных")
        return alerts