-- TEMPLATE: Gaps & Islands
-- Find all consecutive date sequences per user/entity

WITH activity AS (
    -- Step 1: Get one row per user per active date (deduplicate if needed)
    SELECT DISTINCT
        user_id,
        activity_date
    FROM events
),

numbered AS (
    -- Step 2: Assign row number PER USER ordered by date
    SELECT
        user_id,
        activity_date,
        ROW_NUMBER() OVER (
            PARTITION BY user_id
            ORDER BY activity_date
        ) AS rn
    FROM activity
),

islands AS (
    -- Step 3: The KEY TRICK — subtract row number from date
    -- Consecutive dates → same (date - rn) value = same island group
    SELECT
        user_id,
        activity_date,
        rn,
        -- This creates a constant "group key" for each consecutive run:
        (activity_date - INTERVAL '1 day' * (rn - 1)) AS island_group
        -- In Hive: DATE_SUB(activity_date, rn - 1) AS island_group
        -- In Spark: DATE_SUB(activity_date, rn - 1) AS island_group
    FROM numbered
)

-- Step 4: Aggregate each island
SELECT
    user_id,
    MIN(activity_date) AS streak_start,
    MAX(activity_date) AS streak_end,
    COUNT(*) AS streak_length
FROM islands
GROUP BY user_id, island_group
ORDER BY user_id, streak_start;

WITH deduplicated AS (
    -- Remove duplicate logins on same day for same user
    SELECT DISTINCT user_id, login_date
    FROM logins
),

numbered AS (
    SELECT
        user_id,
        login_date,
        ROW_NUMBER() OVER (
            PARTITION BY user_id
            ORDER BY login_date
        ) AS rn
    FROM deduplicated
),

islands AS (
    SELECT
        user_id,
        login_date,
        -- THE TRICK: consecutive dates minus sequential row numbers = constant
        (login_date - CAST(rn AS INT)) AS island_key
        -- Postgres: login_date - rn  (integer subtraction from date)
        -- Hive/Spark: DATE_SUB(login_date, rn)
    FROM numbered
),

streaks AS (
    SELECT
        user_id,
        MIN(login_date) AS streak_start,
        MAX(login_date) AS streak_end,
        COUNT(*) AS streak_length
    FROM islands
    GROUP BY user_id, island_key
),

ranked_streaks AS (
    SELECT
        user_id,
        streak_start,
        streak_end,
        streak_length,
        RANK() OVER (ORDER BY streak_length DESC) AS streak_rank
    FROM streaks
)

SELECT user_id, streak_start, streak_end, streak_length
FROM ranked_streaks
WHERE streak_rank <= 3
ORDER BY streak_length DESC;

WITH price_direction AS (
    SELECT
        stock_symbol,
        price_date,
        closing_price,
        LAG(closing_price) OVER (
            PARTITION BY stock_symbol
            ORDER BY price_date
        ) AS prev_price,
        -- Flag: 1 if price increased from yesterday, 0 otherwise
        CASE WHEN closing_price > LAG(closing_price) OVER (
            PARTITION BY stock_symbol ORDER BY price_date
        ) THEN 1 ELSE 0 END AS is_increase
    FROM stock_prices
),

increases_only AS (
    -- Keep only the increasing-price rows
    SELECT stock_symbol, price_date
    FROM price_direction
    WHERE is_increase = 1
),

numbered AS (
    SELECT
        stock_symbol,
        price_date,
        ROW_NUMBER() OVER (PARTITION BY stock_symbol ORDER BY price_date) AS rn
    FROM increases_only
),

islands AS (
    SELECT
        stock_symbol,
        price_date,
        (price_date - CAST(rn AS INT)) AS island_key
    FROM numbered
),

streaks AS (
    SELECT
        stock_symbol,
        MIN(price_date) AS streak_start,
        MAX(price_date) AS streak_end,
        COUNT(*) AS streak_length
    FROM islands
    GROUP BY stock_symbol, island_key
)

SELECT stock_symbol, streak_start, streak_end, streak_length
FROM streaks
WHERE streak_length >= 3
ORDER BY streak_length DESC;

-- TEMPLATE: Sessionization (create session_id from events)

WITH events_with_prev AS (
    SELECT
        user_id,
        event_time,
        event_type,
        -- Get the previous event time for this user
        LAG(event_time) OVER (
            PARTITION BY user_id
            ORDER BY event_time
        ) AS prev_event_time
    FROM events
),

session_flags AS (
    SELECT
        user_id,
        event_time,
        event_type,
        -- Flag: 1 = start of a NEW session (gap > 30 min OR first event)
        CASE
            WHEN prev_event_time IS NULL THEN 1                       -- first event ever
            WHEN event_time - prev_event_time > INTERVAL '30 minutes' THEN 1  -- gap > threshold
            ELSE 0
        END AS is_new_session
    FROM events_with_prev
),

sessions AS (
    SELECT
        user_id,
        event_time,
        event_type,
        is_new_session,
        -- Cumulative sum of new_session flags = session_id per user!
        SUM(is_new_session) OVER (
            PARTITION BY user_id
            ORDER BY event_time
            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        ) AS session_id
    FROM session_flags
)

-- Now you have session_id on every event. Use it for aggregation:
SELECT
    user_id,
    session_id,
    MIN(event_time) AS session_start,
    MAX(event_time) AS session_end,
    COUNT(*) AS events_in_session,
    MAX(event_time) - MIN(event_time) AS session_duration
FROM sessions
GROUP BY user_id, session_id
ORDER BY user_id, session_id;

WITH events_ordered AS (
    SELECT
        user_id,
        event_id,
        event_time,
        event_type,
        LAG(event_time) OVER (
            PARTITION BY user_id
            ORDER BY event_time
        ) AS prev_event_time
    FROM user_events
),

new_session_flags AS (
    SELECT
        user_id,
        event_id,
        event_time,
        event_type,
        CASE
            WHEN prev_event_time IS NULL THEN 1
            WHEN EXTRACT(EPOCH FROM (event_time - prev_event_time)) / 60 > 30 THEN 1
            -- Hive/Spark: (UNIX_TIMESTAMP(event_time) - UNIX_TIMESTAMP(prev_event_time)) / 60 > 30
            ELSE 0
        END AS new_session
    FROM events_ordered
)

SELECT
    user_id,
    event_id,
    event_time,
    event_type,
    SUM(new_session) OVER (
        PARTITION BY user_id
        ORDER BY event_time
        ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
    ) AS session_id   -- starts at 1 for first session, increments at each new session
FROM new_session_flags
ORDER BY user_id, event_time;

-- TEMPLATE: Deduplication — keep most recent record per key
WITH ranked AS (
    SELECT
        *,
        ROW_NUMBER() OVER (
            PARTITION BY unique_key_col     -- "what defines a duplicate?"
            ORDER BY updated_at DESC        -- "which one to keep?" (DESC = latest first)
        ) AS rn
    FROM source_table
)

SELECT * EXCEPT (rn)   -- or SELECT col1, col2... (exclude the rn column)
FROM ranked
WHERE rn = 1;          -- keep only the first row (= latest record) per key

-- ⚠️ ROW_NUMBER always gives unique ranks (no ties)
-- → exactly 1 row per partition will have rn = 1
-- ⚠️ RANK() could give multiple rows with rank=1 if tie on updated_at → unsafe for dedup

WITH latest_record AS (
    SELECT
        customer_id,
        name,
        email,
        address,
        modified_at,
        ROW_NUMBER() OVER (
            PARTITION BY customer_id
            ORDER BY modified_at DESC    -- latest timestamp first → rn=1 is most recent
        ) AS rn
    FROM customer_cdc
)

SELECT customer_id, name, email, address, modified_at
FROM latest_record
WHERE rn = 1;

-- ⚠️ In production Databricks/Delta Lake:
-- MERGE INTO customer_dim USING (SELECT ... WHERE rn=1) ON customer_id MATCH
-- This is the SCD Type 1 (overwrite) pattern

-- Approach 1: Find the emails that have duplicates
SELECT
    email,
    COUNT(DISTINCT user_id) AS account_count
FROM users
GROUP BY email
HAVING COUNT(DISTINCT user_id) > 1;

-- Approach 2: Return ALL user_ids that share an email with another user
SELECT u.user_id, u.email
FROM users u
WHERE u.email IN (
    SELECT email
    FROM users
    GROUP BY email
    HAVING COUNT(DISTINCT user_id) > 1
)
ORDER BY u.email, u.user_id;