스키마 진화와 스키마 강제 기초

-- 테이블 정의
CREATE TABLE catalog.schema.products (
    product_id BIGINT,
    name STRING,
    price DECIMAL(10, 2)
);

-- ✅ 정상: 스키마가 일치
INSERT INTO catalog.schema.products VALUES (1, '노트북', 1500000.00);

-- ❌ 에러: 'category' 컬럼이 테이블에 없음
INSERT INTO catalog.schema.products (product_id, name, price, category)
VALUES (2, '키보드', 89000.00, '전자제품');
-- AnalysisException: Cannot write to table because the column 'category' is not found

-- ❌ 에러: price에 문자열 삽입 시도
INSERT INTO catalog.schema.products VALUES (3, '마우스', 'cheap');
-- AnalysisException: Cannot safely cast 'price': string to decimal(10,2)

-- 컬럼 추가
ALTER TABLE catalog.schema.products
ADD COLUMN category STRING;

-- 컬럼 추가 (기본값 포함)
ALTER TABLE catalog.schema.products
ADD COLUMN created_at TIMESTAMP DEFAULT current_timestamp();

-- 컬럼 이름 변경 (Column Mapping 필요)
ALTER TABLE catalog.schema.products
RENAME COLUMN name TO product_name;

-- 컬럼 삭제 (Column Mapping 필요)
ALTER TABLE catalog.schema.products
DROP COLUMN category;

-- 컬럼 타입 변경 (안전한 확장만 가능)
ALTER TABLE catalog.schema.products
ALTER COLUMN product_id TYPE BIGINT;  -- INT → BIGINT은 가능

-- SQL: INSERT 시 mergeSchema 활성화
SET spark.databricks.delta.schema.autoMerge.enabled = true;

INSERT INTO catalog.schema.products VALUES
(4, '태블릿', 800000.00, '전자제품', current_timestamp());
-- 'category' 컬럼이 자동으로 추가됩니다

# PySpark: DataFrame 쓰기 시 mergeSchema 옵션
df_new = spark.createDataFrame([
    (5, "스피커", 250000.00, "전자제품", "2025-03-20")
], ["product_id", "name", "price", "category", "launch_date"])

df_new.write \
    .mode("append") \
    .option("mergeSchema", "true") \
    .saveAsTable("catalog.schema.products")

-- MERGE에서도 스키마 진화가 가능합니다
SET spark.databricks.delta.schema.autoMerge.enabled = true;

MERGE INTO catalog.schema.products AS target
USING new_products AS source
ON target.product_id = source.product_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;
-- source에 새 컬럼이 있으면 자동으로 추가됩니다

-- Column Mapping 활성화
ALTER TABLE catalog.schema.products
SET TBLPROPERTIES ('delta.columnMapping.mode' = 'name');

-- Column Mapping이 활성화된 테이블에서

-- 컬럼 이름 변경
ALTER TABLE catalog.schema.products
RENAME COLUMN name TO product_name;

-- 컬럼 삭제
ALTER TABLE catalog.schema.products
DROP COLUMN old_unused_column;

-- 여러 컬럼 동시 삭제
ALTER TABLE catalog.schema.products
DROP COLUMNS (temp_col1, temp_col2);

# Auto Loader에서 스키마 진화 활성화
(spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", "/checkpoints/schema")
    # 스키마 진화 모드 설정
    .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
    .load("s3://my-bucket/raw-data/")
    .writeStream
    .option("checkpointLocation", "/checkpoints/stream")
    .option("mergeSchema", "true")
    .table("catalog.schema.bronze_events")
)