Schema Evolution: стратегии добавления и удаления колонок без поломки

from pyspark.sql import SparkSession
from pyspark.sql.types import (
    StructType, StructField, StringType, DecimalType, TimestampType
)
from pyspark.sql import functions as F

spark = SparkSession.builder \
    .appName("Schema-Evolution") \
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.lakehouse", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.lakehouse.type", "hadoop") \
    .config("spark.sql.catalog.lakehouse.warehouse", "/tmp/warehouse") \
    .getOrCreate()

# Создаём Silver-таблицу с фиксированной схемой
spark.sql("""
    CREATE TABLE IF NOT EXISTS lakehouse.silver.orders (
        order_id    INTEGER,
        customer_id STRING,
        status      STRING,
        amount      DECIMAL(10,2),
        created_at  TIMESTAMP
    )
    USING iceberg
""")

# Пишем корректные данные - всё работает
correct_df = spark.createDataFrame(
    [(1001, "cust_A", "confirmed", 1500.0, "2026-05-23 10:00:00")],
    ["order_id", "customer_id", "status", "amount", "created_at"]
).withColumn("created_at", F.to_timestamp("created_at")) \
 .withColumn("amount", F.col("amount").cast("decimal(10,2)"))

correct_df.writeTo("lakehouse.silver.orders").append()

# Попытка записать DataFrame с ДОПОЛНИТЕЛЬНОЙ колонкой
df_with_new_col = correct_df.withColumn("discount_code", F.lit("SALE20"))

try:
    df_with_new_col.writeTo("lakehouse.silver.orders").append()
except Exception as e:
    print(f"Schema Enforcement сработал: {type(e).__name__}")
    print(f"Сообщение: {str(e)[:200]}")
    # AnalysisException: Cannot write incompatible data to table 'silver.orders'
    # Found column discount_code which doesn't match target table schema

# Попытка записать с ДРУГИМ ТИПОМ
df_wrong_type = spark.createDataFrame(
    [(1002, "cust_B", "pending", "NOT_A_NUMBER", "2026-05-23 11:00:00")],
    ["order_id", "customer_id", "status", "amount", "created_at"]
)

try:
    df_wrong_type.writeTo("lakehouse.silver.orders").append()
except Exception as e:
    print(f"Type mismatch: {type(e).__name__}")

# Сценарий: PostgreSQL добавил колонку discount_code в таблицу orders
# Debezium начинает присылать её в CDC-потоке

# Первый батч (до изменения схемы): 3 поля
batch_v1 = spark.createDataFrame([
    (1001, "cust_A", "confirmed", 1500.0, "2026-05-23 10:00:00", None),
    (1002, "cust_B", "pending",    750.0, "2026-05-23 11:00:00", None),
], ["order_id", "customer_id", "status", "amount", "created_at", "discount_code"])

batch_v1.writeTo("lakehouse.bronze.orders").append()

# Второй батч (ПОСЛЕ изменения схемы): новое поле discount_code
batch_v2 = spark.createDataFrame([
    (1003, "cust_C", "confirmed", 2000.0, "2026-05-23 12:00:00", "SALE20"),
    (1004, "cust_D", "cancelled",  500.0, "2026-05-23 13:00:00", "VIP10"),
], ["order_id", "customer_id", "status", "amount", "created_at", "discount_code"])

# Без mergeSchema: AnalysisException - discount_code не в схеме таблицы
try:
    batch_v2.writeTo("lakehouse.bronze.orders").append()
except Exception as e:
    print("Ожидаемая ошибка без mergeSchema:", type(e).__name__)

# С mergeSchema: Iceberg добавляет колонку в schema таблицы атомарно
batch_v2.writeTo("lakehouse.bronze.orders") \
    .option("mergeSchema", "true") \
    .append()

print("Schema после добавления колонки:")
spark.table("lakehouse.bronze.orders").printSchema()
# root
#  |-- order_id:      integer
#  |-- customer_id:   string
#  |-- status:        string
#  |-- amount:        decimal(10,2)
#  |-- created_at:    string
#  |-- discount_code: string    ← добавлена автоматически

# Демонстрация backward-compatible чтения
all_orders = spark.table("lakehouse.bronze.orders")

print("Строки ДО добавления поля:")
all_orders.filter(F.col("order_id").isin(1001, 1002)) \
    .select("order_id", "discount_code") \
    .show()
# order_id | discount_code
# 1001     | null          ← NULL для старых строк
# 1002     | null

print("Строки ПОСЛЕ добавления поля:")
all_orders.filter(F.col("order_id").isin(1003, 1004)) \
    .select("order_id", "discount_code") \
    .show()
# order_id | discount_code
# 1003     | SALE20
# 1004     | VIP10

# Явное добавление колонки через DDL
# Плюс: полный контроль над типом, позицией, комментарием
# Плюс: изменение задокументировано в DDL-скрипте (git history)
spark.sql("""
    ALTER TABLE lakehouse.silver.orders
    ADD COLUMN discount_code STRING COMMENT 'Промо-код, применённый к заказу (nullable)'
""")

# Добавление нескольких колонок одновременно
spark.sql("""
    ALTER TABLE lakehouse.silver.orders
    ADD COLUMNS (
        discount_amount DECIMAL(8,2) COMMENT 'Сумма скидки в рублях',
        is_first_order  BOOLEAN      COMMENT 'Флаг первого заказа клиента'
    )
""")

# Проверяем обновлённую схему
spark.sql("DESCRIBE TABLE lakehouse.silver.orders").show(20)

from pyspark.sql.types import StructType, StructField, StringType, DecimalType

# Исходная схема: metadata как struct
schema_v1 = StructType([
    StructField("order_id",   StringType(), True),
    StructField("metadata",   StructType([
        StructField("source",   StringType(), True),
        StructField("version",  StringType(), True),
    ]), True),
])

# Новая версия: metadata получил дополнительное поле experiment_id
schema_v2 = StructType([
    StructField("order_id",   StringType(), True),
    StructField("metadata",   StructType([
        StructField("source",        StringType(), True),
        StructField("version",       StringType(), True),
        StructField("experiment_id", StringType(), True),  # новое поле
    ]), True),
])

data_v2 = spark.createDataFrame([
    ("order_1005", {"source": "mobile", "version": "3.1", "experiment_id": "exp-42"}),
], schema_v2)

# mergeSchema работает и для nested struct
data_v2.writeTo("lakehouse.bronze.orders_nested") \
    .option("mergeSchema", "true") \
    .append()

# Чтение: старые строки получат metadata.experiment_id = null
# Новые строки - реальное значение
spark.table("lakehouse.bronze.orders_nested") \
    .select("order_id", "metadata.source", "metadata.experiment_id") \
    .show()

# Ситуация: команда решила убрать deprecated поле user_id (заменили на customer_id)
# Сначала проверяем, кто зависит от этой колонки

# 1. Проверяем Silver jobs - используют ли они user_id?
# 2. Проверяем Gold marts - есть ли JOIN по user_id?
# 3. Проверяем BI-дашборды - есть ли SELECT user_id?

# Только после аудита зависимостей - физическое удаление
spark.sql("""
    ALTER TABLE lakehouse.silver.orders DROP COLUMN user_id
""")

# ПРОБЛЕМА: физическое удаление в Iceberg (Copy-on-Write)
# перезаписывает ВСЕ data files, из которых нужно убрать колонку
# Для таблицы в 10 ТБ - это долго и дорого

# Включаем Column Mapping для таблицы
spark.sql("""
    ALTER TABLE lakehouse.silver.orders
    SET TBLPROPERTIES (
        'write.metadata.metrics.default' = 'truncate(16)',
        'format-version' = '2'
    )
""")

# Логическое переименование: физический файл хранит "user_id",
# но Iceberg metadata маппит его на "customer_id"
spark.sql("""
    ALTER TABLE lakehouse.silver.orders
    RENAME COLUMN user_id TO customer_id
""")

# Читатели, использующие customer_id, получают корректные данные
# Физические файлы НЕ перезаписываются - операция мгновенная!
# Старые читатели, использующие user_id, получат ошибку (это корректное поведение)

# Фаза 1: Dual Write в Silver pipeline
def transform_to_silver(bronze_df):
    return (
        bronze_df
        .withColumn("customer_id",
                    F.coalesce(F.col("customer_id"), F.col("user_id")))
        .withColumn("user_id",
                    F.col("user_id"))  # Сохраняем старое поле пока
        # ... остальные трансформации
    )

# Фаза 3 (через 2 недели): убираем user_id из pipeline
def transform_to_silver_v2(bronze_df):
    return (
        bronze_df
        .withColumn("customer_id",
                    F.coalesce(F.col("customer_id"), F.col("user_id")))
        .drop("user_id")  # Больше не нужно
    )

# Безопасные расширения типов (Type Widening)
# Iceberg умеет делать их логически, без перезаписи файлов

# INT → BIGINT (расширение диапазона)
spark.sql("""
    ALTER TABLE lakehouse.silver.orders
    ALTER COLUMN order_id TYPE BIGINT
""")
# Старые файлы хранят INT - при чтении Iceberg автоматически преобразует в BIGINT
# Новые файлы записываются как BIGINT

# FLOAT → DOUBLE (расширение точности)
spark.sql("""
    ALTER TABLE lakehouse.silver.orders
    ALTER COLUMN amount TYPE DOUBLE
""")

# Проверяем изменение
spark.table("lakehouse.silver.orders").printSchema()
# order_id: bigint  ← теперь BIGINT
# amount: double    ← теперь DOUBLE

# Bronze pipeline с автоматическим mergeSchema
# Этот код корректно обрабатывает любые schema changes в источнике

def ingest_to_bronze(source_df, target_table: str, pipeline_run_id: str):
    """
    Universal Bronze ingestion с автоматической эволюцией схемы.
    Принимает любые новые поля от источника без ошибок.
    """
    enriched_df = (
        source_df
        .withColumn("_ingested_at",      F.current_timestamp())
        .withColumn("_pipeline_run_id",  F.lit(pipeline_run_id))
        .withColumn("_source_schema_hash",
            # Хэш схемы для отслеживания изменений
            F.sha2(F.lit(str(source_df.schema.simpleString())), 256))
        .withColumn("ingestion_date", F.current_date())
    )

    # mergeSchema=true: любые новые поля из источника автоматически добавятся
    enriched_df.writeTo(target_table) \
        .option("mergeSchema", "true") \
        .append()

    # Логируем schema change если он произошёл
    current_schema = spark.table(target_table).schema
    print(f"Текущая схема {target_table}: {len(current_schema.fields)} колонок")

# Первый запуск: 5 полей
batch1 = spark.createDataFrame([
    ("1001", "cust_A", "confirmed", "1500.0", "2026-05-23"),
], ["order_id", "customer_id", "status", "amount", "event_date"])

ingest_to_bronze(batch1, "lakehouse.bronze.orders", "run-001")
# Схема: 5 + 4 метаданных = 9 колонок

# Второй запуск: источник добавил discount_code и experiment_id
batch2 = spark.createDataFrame([
    ("1002", "cust_B", "pending", "750.0", "2026-05-23", "SALE20", "exp-42"),
], ["order_id", "customer_id", "status", "amount", "event_date",
    "discount_code", "experiment_id"])

ingest_to_bronze(batch2, "lakehouse.bronze.orders", "run-002")
# Схема автоматически расширилась: 9 → 11 колонок
# batch1 строки получат discount_code=NULL, experiment_id=NULL

# Детектирование schema changes для мониторинга и алертинга
def detect_schema_change(spark, bronze_table: str):
    """
    Сравнивает текущую схему с предыдущей.
    Если схема изменилась - логирует предупреждение.
    """
    schema_history = (
        spark.table(bronze_table)
        .groupBy("_source_schema_hash", "ingestion_date")
        .agg(F.count("*").alias("row_count"))
        .orderBy("ingestion_date")
    )

    # Если в одном дне несколько разных хэшей - схема менялась внутри дня
    daily_schemas = (
        schema_history
        .groupBy("ingestion_date")
        .agg(F.countDistinct("_source_schema_hash").alias("schema_versions"))
        .filter(F.col("schema_versions") > 1)
    )

    if daily_schemas.count() > 0:
        print("ВНИМАНИЕ: Schema change обнаружен!")
        daily_schemas.show()

detect_schema_change(spark, "lakehouse.bronze.orders")

# migration_20260523_001_add_discount_fields.py
# Миграция: добавление полей для работы со скидками

import datetime
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

MIGRATION_ID = "20260523_001"
MIGRATION_DESC = "Добавление полей discount_code и discount_amount в silver.orders"

def run_migration(spark):
    """
    Идемпотентная миграция схемы Silver.
    Безопасно запускать несколько раз.
    """
    print(f"Запуск миграции {MIGRATION_ID}: {MIGRATION_DESC}")

    # Проверяем: миграция уже выполнялась?
    existing_columns = [
        f.name for f in spark.table("lakehouse.silver.orders").schema.fields
    ]

    if "discount_code" in existing_columns:
        print(f"Миграция {MIGRATION_ID} уже применена. Пропускаем.")
        return

    # Шаг 1: Добавляем новые колонки
    spark.sql("""
        ALTER TABLE lakehouse.silver.orders
        ADD COLUMNS (
            discount_code   STRING      COMMENT 'Промо-код из источника (nullable)',
            discount_amount DECIMAL(8,2) COMMENT 'Сумма скидки в рублях (nullable)'
        )
    """)
    print("  Шаг 1/3: Колонки добавлены в схему таблицы")

    # Шаг 2: Бэкфилл - заполняем discount_amount для исторических строк
    # Если у нас есть данные в Bronze, можем пересчитать
    # Если нет - оставляем NULL (backward compatible)
    print("  Шаг 2/3: Исторические данные оставляем NULL (backward compatible)")

    # Шаг 3: Валидируем результат
    schema_after = spark.table("lakehouse.silver.orders").schema
    assert "discount_code" in [f.name for f in schema_after.fields], \
        "Миграция не применилась!"
    print("  Шаг 3/3: Валидация прошла")

    # Логируем факт миграции
    spark.sql(f"""
        INSERT INTO lakehouse.metadata.schema_migrations
        (migration_id, description, applied_at, status)
        VALUES ('{MIGRATION_ID}', '{MIGRATION_DESC}', current_timestamp(), 'SUCCESS')
    """)

    print(f"Миграция {MIGRATION_ID} успешно применена.")


spark = SparkSession.builder.appName("Migration").getOrCreate()
run_migration(spark)

from pyspark.sql.types import (
    StructType, StructField, StringType, DecimalType, TimestampType
)

# Определяем «максимальную» схему - включает поля всех известных версий
# Все поля, которые могут отсутствовать - nullable
TARGET_SCHEMA = StructType([
    StructField("order_id",      StringType(),         False),  # обязательное
    StructField("status",        StringType(),         True),
    StructField("amount",        DecimalType(10, 2),   True),
    StructField("event_ts",      TimestampType(),      True),
    StructField("discount_code", StringType(),         True),   # v2 поле, nullable
    StructField("experiment_id", StringType(),         True),   # v3 поле, nullable
])

# Читаем JSON из Kafka, применяем целевую схему
# from_json заполняет отсутствующие поля NULL автоматически
streaming_df = (
    spark.readStream
         .format("kafka")
         .option("kafka.bootstrap.servers", "kafka:9092")
         .option("subscribe", "orders-events")
         .load()
         .select(
             F.from_json(
                 F.col("value").cast("string"),
                 TARGET_SCHEMA
             ).alias("data")
         )
         .select("data.*")
         # Событие v1: discount_code=NULL, experiment_id=NULL
         # Событие v2: discount_code="SALE20", experiment_id=NULL
         # Событие v3: discount_code="SALE20", experiment_id="exp-42"
)

# Запись в Bronze Iceberg с mergeSchema для будущих версий
query = (
    streaming_df
    .writeStream
    .format("iceberg")
    .outputMode("append")
    .option("checkpointLocation", "s3://checkpoints/orders-events/")
    .option("fanout-enabled", "true")
    # mergeSchema позволяет добавить v4 поля без остановки стрима
    .option("mergeSchema", "true")
    .toTable("lakehouse.bronze.order_events")
)

# АНТИПАТТЕРН: глобальное включение autoMerge без контроля

# В SparkSession config или SQL:
spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", "true")
# Теперь ЛЮБАЯ запись в ЛЮБУЮ таблицу автоматически расширяет схему

# Проблема: разработчик случайно пишет DataFrame с колонкой-опечаткой
bad_df = spark.createDataFrame([
    (1001, "cust_A", "confirmed", 1500.0, "SALE20")
], ["order_id", "customer_id", "status", "amount", "discont_code"])
# ^ опечатка: "discont_code" вместо "discount_code"

bad_df.writeTo("lakehouse.silver.orders").append()
# Без контроля: в таблице появится и "discount_code" и "discont_code"
# Silver-таблица молча загрязнена. Аналитики увидят дубликат колонки.

# АНТИПАТТЕРН: schema inference из данных в production

# Плохо: Spark пытается вывести схему из первых строк файла
df_inferred = spark.read.json("s3://raw/orders/2026-05-23.json")

# Проблемы:
# 1. Схема выводится из ПОДВЫБОРКИ данных - в конце файла могут быть поля,
#    которых нет в начале. Spark не увидит их.
# 2. Типы выводятся эвристически: "1500" → LONG, не DECIMAL
# 3. Опциональные поля могут вывестись как NOT NULL если в sample нет null
# 4. При следующем запуске inference может дать другую схему!

# Хорошо: явная схема
EXPLICIT_SCHEMA = StructType([
    StructField("order_id",     StringType(),        False),
    StructField("customer_id",  StringType(),        True),
    StructField("status",       StringType(),        True),
    StructField("amount",       DecimalType(10, 2),  True),
    StructField("created_at",   TimestampType(),     True),
    StructField("discount_code", StringType(),       True),
])

df_explicit = spark.read.schema(EXPLICIT_SCHEMA).json("s3://raw/orders/2026-05-23.json")
# Схема детерминирована: тот же результат при каждом запуске

# Правильный подход: автоматическое обнаружение schema changes

def validate_schema_unchanged(spark, table: str, expected_schema: StructType):
    """
    Проверяет, что схема таблицы не изменилась.
    Запускается как шаг pipeline перед обработкой.
    """
    current_schema = spark.table(table).schema
    current_fields = {f.name: f.dataType for f in current_schema.fields}
    expected_fields = {f.name: f.dataType for f in expected_schema.fields}

    # Новые поля (добавленные без нашего ведома)
    new_fields = set(current_fields.keys()) - set(expected_fields.keys())
    # Удалённые поля
    removed_fields = set(expected_fields.keys()) - set(current_fields.keys())
    # Изменённые типы
    type_changes = {
        name: (expected_fields[name], current_fields[name])
        for name in expected_fields
        if name in current_fields and expected_fields[name] != current_fields[name]
    }

    if new_fields:
        print(f"ПРЕДУПРЕЖДЕНИЕ: Обнаружены новые поля в {table}: {new_fields}")
    if removed_fields:
        raise ValueError(f"КРИТИЧНО: Удалены поля из {table}: {removed_fields}")
    if type_changes:
        raise ValueError(f"КРИТИЧНО: Изменены типы в {table}: {type_changes}")

    print(f"✓ Схема {table} соответствует ожидаемой")

import datetime
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.window import Window

spark = SparkSession.builder \
    .appName("Schema-Evolution-E2E") \
    .config("spark.sql.extensions",
            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.lakehouse", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.lakehouse.type", "hadoop") \
    .config("spark.sql.catalog.lakehouse.warehouse", "/tmp/warehouse") \
    .getOrCreate()

# ─── СОСТОЯНИЕ ДО ИЗМЕНЕНИЯ ──────────────────────────────────────────────────
print("=== ФАЗА 0: Исходное состояние ===")

# Bronze: 5 колонок из старого CDC
spark.sql("""
    CREATE TABLE IF NOT EXISTS lakehouse.bronze.orders_evo (
        order_id     STRING,
        customer_id  STRING,
        status       STRING,
        amount       STRING,
        created_at   STRING,
        _ingested_at  TIMESTAMP,
        ingestion_date DATE
    )
    USING iceberg
    PARTITIONED BY (ingestion_date)
""")

old_data = spark.createDataFrame([
    ("1001", "cust_A", "confirmed", "1500.0", "2026-05-23 10:00:00"),
    ("1002", "cust_B", "pending",    "750.0", "2026-05-23 11:00:00"),
], ["order_id", "customer_id", "status", "amount", "created_at"])

(old_data
 .withColumn("_ingested_at", F.current_timestamp())
 .withColumn("ingestion_date", F.current_date())
 .writeTo("lakehouse.bronze.orders_evo").append())

print("Bronze содержит:", spark.table("lakehouse.bronze.orders_evo").count(), "строк")

# Silver: типизированные данные без discount_code
spark.sql("""
    CREATE TABLE IF NOT EXISTS lakehouse.silver.orders_evo (
        order_id     INTEGER,
        customer_id  STRING,
        status       STRING,
        amount       DECIMAL(10,2),
        created_at   TIMESTAMP
    )
    USING iceberg
    PARTITIONED BY (month(created_at))
""")

old_silver = (
    spark.table("lakehouse.bronze.orders_evo")
    .withColumn("order_id",   F.col("order_id").cast("integer"))
    .withColumn("amount",     F.col("amount").cast("decimal(10,2)"))
    .withColumn("created_at", F.to_timestamp("created_at"))
    .select("order_id", "customer_id", "status", "amount", "created_at")
)
old_silver.writeTo("lakehouse.silver.orders_evo").append()

# Gold: агрегат без скидок
spark.sql("""
    CREATE TABLE IF NOT EXISTS lakehouse.gold.revenue_evo (
        order_date    DATE,
        total_revenue DECIMAL(15,2),
        _updated_at   TIMESTAMP
    )
    USING iceberg
    PARTITIONED BY (order_date)
""")

(spark.table("lakehouse.silver.orders_evo")
 .filter(F.col("status") == "confirmed")
 .groupBy(F.to_date("created_at").alias("order_date"))
 .agg(F.sum("amount").alias("total_revenue"))
 .withColumn("_updated_at", F.current_timestamp())
 .writeTo("lakehouse.gold.revenue_evo").overwritePartitions())

print(f"Исходная схема Silver: "
      f"{[f.name for f in spark.table('lakehouse.silver.orders_evo').schema.fields]}")

# ─── ФАЗА 1: BRONZE ПРИНИМАЕТ НОВОЕ ПОЛЕ ─────────────────────────────────────
print("\n=== ФАЗА 1: Bronze принимает discount_code ===")

new_cdc_data = spark.createDataFrame([
    ("1003", "cust_C", "confirmed", "2000.0", "2026-05-23 12:00:00", "SALE20"),
    ("1004", "cust_D", "cancelled", "500.0",  "2026-05-23 13:00:00", "VIP10"),
], ["order_id", "customer_id", "status", "amount", "created_at", "discount_code"])

# mergeSchema: Bronze автоматически принимает новое поле
(new_cdc_data
 .withColumn("_ingested_at", F.current_timestamp())
 .withColumn("ingestion_date", F.current_date())
 .writeTo("lakehouse.bronze.orders_evo")
 .option("mergeSchema", "true")
 .append())

print("Bronze после Schema Evolution:")
spark.table("lakehouse.bronze.orders_evo") \
     .select("order_id", "discount_code") \
     .show()
# Старые строки: discount_code=NULL
# Новые строки:  discount_code="SALE20" / "VIP10"

# ─── ФАЗА 2: МИГРАЦИЯ SILVER ──────────────────────────────────────────────────
print("\n=== ФАЗА 2: Явная миграция Silver ===")

# Шаг 2а: DDL - добавляем колонку в Silver явно
spark.sql("""
    ALTER TABLE lakehouse.silver.orders_evo
    ADD COLUMN discount_code STRING COMMENT 'Промо-код (nullable, добавлен 2026-05-23)'
""")

# Шаг 2б: Инкрементальное обновление - MERGE в Silver с новым полем
new_silver = (
    spark.table("lakehouse.bronze.orders_evo")
    .filter(F.col("ingestion_date") == F.current_date())
    .filter(F.col("order_id").isin("1003", "1004"))
    .withColumn("order_id",   F.col("order_id").cast("integer"))
    .withColumn("amount",     F.col("amount").cast("decimal(10,2)"))
    .withColumn("created_at", F.to_timestamp("created_at"))
    .select("order_id", "customer_id", "status", "amount",
            "created_at", "discount_code")
)

new_silver.createOrReplaceTempView("new_silver_orders")

spark.sql("""
    MERGE INTO lakehouse.silver.orders_evo AS target
    USING new_silver_orders AS source
    ON target.order_id = source.order_id
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED THEN INSERT *
""")

print("Silver после миграции:")
spark.table("lakehouse.silver.orders_evo").show()
# Старые строки: discount_code=NULL (backward compatible)
# Новые строки: discount_code="SALE20" / "VIP10"

# ─── ФАЗА 3: GOLD РАСШИРЯЕТСЯ ─────────────────────────────────────────────────
print("\n=== ФАЗА 3: Gold с метрикой скидок ===")

# Gold: добавляем метрику суммы скидок
spark.sql("""
    ALTER TABLE lakehouse.gold.revenue_evo
    ADD COLUMN discounted_orders_count BIGINT
        COMMENT 'Число заказов с промо-кодом (добавлено 2026-05-23)'
""")

# Пересчитываем Gold с новой метрикой
updated_gold = (
    spark.table("lakehouse.silver.orders_evo")
    .filter(F.col("status") == "confirmed")
    .groupBy(F.to_date("created_at").alias("order_date"))
    .agg(
        F.sum("amount").alias("total_revenue"),
        # Новая метрика: сколько заказов с промо-кодом
        F.count(
            F.when(F.col("discount_code").isNotNull(), 1)
        ).alias("discounted_orders_count"),
    )
    .withColumn("_updated_at", F.current_timestamp())
)

updated_gold.writeTo("lakehouse.gold.revenue_evo").overwritePartitions()

print("Gold после эволюции схемы:")
spark.table("lakehouse.gold.revenue_evo").show()
# order_date  | total_revenue | discounted_orders_count | _updated_at
# 2026-05-23  | 3500.00       | 1                       | ...
# Старые BI-запросы на total_revenue не сломались!
# Новые дашборды могут использовать discounted_orders_count

# ─── ФИНАЛ: ПРОВЕРКА СОВМЕСТИМОСТИ ───────────────────────────────────────────
print("\n=== ФИНАЛ: Проверка backward compatibility ===")

# «Старый» BI-запрос (SELECT без новых полей) работает без изменений
old_bi_query = spark.sql("""
    SELECT order_date, total_revenue
    FROM lakehouse.gold.revenue_evo
    ORDER BY order_date
""")
old_bi_query.show()
print("Старые BI-запросы работают без изменений ✓")

=== ФАЗА 0: Исходное состояние ===
Bronze содержит: 2 строки
Исходная схема Silver: ['order_id', 'customer_id', 'status', 'amount', 'created_at']

=== ФАЗА 1: Bronze принимает discount_code ===
Bronze после Schema Evolution:
order_id | discount_code
1001     | null           ← backward compatible
1002     | null
1003     | SALE20
1004     | VIP10

=== ФАЗА 2: Явная миграция Silver ===
Silver после миграции:
order_id | status    | amount | discount_code
1001     | confirmed | 1500.0 | null          ← старые строки не тронуты
1002     | pending   | 750.0  | null
1003     | confirmed | 2000.0 | SALE20
1004     | cancelled | 500.0  | VIP10

=== ФАЗА 3: Gold с метрикой скидок ===
Gold после эволюции:
order_date  | total_revenue | discounted_orders_count
2026-05-23  | 3500.00       | 1

=== ФИНАЛ ===
Старые BI-запросы работают без изменений ✓