# Python + SQLAlchemy: Watermark-basierter inkrementeller Load
from datetime import datetime
import sqlalchemy as sa

def get_last_watermark(engine, pipeline_name: str) -> datetime:
    # Letzten erfolgreichen Watermark aus Control-Tabelle holen
    with engine.connect() as conn:
        result = conn.execute(
            sa.text(
                "SELECT last_watermark FROM pipeline_state WHERE pipeline_name = :name"
            ),
            {"name": pipeline_name}
        ).fetchone()
    return result[0] if result else datetime(2000, 1, 1)  # Default für ersten Lauf

def set_watermark(engine, pipeline_name: str, new_watermark: datetime):
    # Watermark nach erfolgreichem Load updaten
    with engine.connect() as conn:
        conn.execute(
            sa.text(
                "INSERT INTO pipeline_state (pipeline_name, last_watermark) "
                "VALUES (:name, :wm) "
                "ON CONFLICT (pipeline_name) DO UPDATE SET last_watermark = EXCLUDED.last_watermark"
            ),
            {"name": pipeline_name, "wm": new_watermark}
        )
        conn.commit()

# Main inkrementeller Load
engine = sa.create_engine("postgresql://...")
last_wm = get_last_watermark(engine, "orders_pipeline")

with engine.connect() as conn:
    new_rows = conn.execute(
        sa.text(
            "SELECT * FROM source.orders WHERE updated_at > :last_wm ORDER BY updated_at"
        ),
        {"last_wm": last_wm}
    ).fetchall()

max_ts = max(row.updated_at for row in new_rows) if new_rows else last_wm
# ... new_rows ins Target laden ...
set_watermark(engine, "orders_pipeline", max_ts)

-- dbt (Spark SQL): Incremental-Model mit Watermark-Strategie
{{ config(
    materialized='incremental',
    incremental_strategy='insert_overwrite',
    partition_by={'field': 'order_date', 'data_type': 'date'}
) }}

SELECT
    order_id,
    customer_id,
    amount,
    order_date,
    updated_at
FROM {{ source('raw', 'orders') }}

{% if is_incremental() %}
WHERE updated_at > (SELECT MAX(updated_at) FROM {{ this }})
{% endif %}

-- Spark SQL: MERGE-Statement für inkrementellen Upsert
MERGE INTO target.dim_customer AS target
USING (
    SELECT
        customer_id,
        customer_name,
        email,
        city,
        updated_at
    FROM source.customers
    WHERE updated_at > '2024-01-01T00:00:00Z'  -- Lookback-Window
) AS source
ON target.customer_id = source.customer_id

WHEN MATCHED AND source.updated_at > target.updated_at THEN
    UPDATE SET
        target.customer_name = source.customer_name,
        target.email         = source.email,
        target.city          = source.city,
        target.updated_at    = source.updated_at

WHEN NOT MATCHED THEN
    INSERT (customer_id, customer_name, email, city, updated_at)
    VALUES (source.customer_id, source.customer_name, source.email, source.city, source.updated_at)

# Python: Merge mit 3-Tage-Lookback-Window
from datetime import datetime, timedelta, timezone

lookback_days = 3
lookback_start = datetime.now(timezone.utc) - timedelta(days=lookback_days)

query = (
    "MERGE INTO target.fct_events AS t "
    "USING ( "
    "    SELECT event_id, user_id, event_type, event_ts "
    "    FROM source.events "
    f"    WHERE event_ts >= '{lookback_start.isoformat()}' "
    ") AS s "
    "ON t.event_id = s.event_id "
    "WHEN MATCHED THEN UPDATE SET t.event_type = s.event_type, t.event_ts = s.event_ts "
    "WHEN NOT MATCHED THEN INSERT (event_id, user_id, event_type, event_ts) "
    "VALUES (s.event_id, s.user_id, s.event_type, s.event_ts)"
)

# Python + PySpark: Append-only inkrementeller Load
from pyspark.sql import SparkSession
from pyspark.sql.functions import current_timestamp

spark = SparkSession.builder.getOrCreate()

# Nur neue Events aus der Source lesen (Kafka, Log-Files, etc.)
new_events = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "user_events")
    .load()
)

# An Delta-Lake-Tabelle anhängen
(
    new_events
    .withColumn("ingested_at", current_timestamp())
    .writeStream
    .format("delta")
    .option("checkpointLocation", "/checkpoints/events")
    .outputMode("append")
    .start("s3://my-bucket/events/")
)

-- Spark SQL: aktuellen Zustand aus Append-only Event-Log holen
-- ROW_NUMBER für die neueste Version jedes Customer-Records
WITH ranked AS (
    SELECT
        customer_id,
        customer_name,
        email,
        city,
        event_ts,
        ROW_NUMBER() OVER (
            PARTITION BY customer_id
            ORDER BY event_ts DESC
        ) AS rn
    FROM events.customer_updates
)
SELECT customer_id, customer_name, email, city
FROM ranked
WHERE rn = 1

# Python + PySpark: idempotenter Append mit Deduplizierung
from pyspark.sql.functions import col

# Eingehenden Batch lesen
incoming = spark.read.parquet("s3://landing/events/2024/03/15/")

# Existierende Keys aus Target lesen (nur Key-Spalte — kein Full-Scan)
existing_keys = (
    spark.read
    .format("delta")
    .load("s3://target/events/")
    .select("event_id")
)

# Nur wirklich neue Zeilen behalten
new_only = incoming.join(existing_keys, on="event_id", how="left_anti")

# Anhängen
(
    new_only.write
    .format("delta")
    .mode("append")
    .save("s3://target/events/")
)