-- The old way: ugly, has race conditions, requires two round-trips for the ID
INSERT INTO events (user_id, event_type, payload, created_at)
VALUES ($1, $2, $3, NOW())
ON CONFLICT (user_id, event_type)
DO UPDATE SET
    payload = EXCLUDED.payload,
    updated_at = NOW()
RETURNING id, created_at, (xmax = 0) AS inserted;

MERGE INTO user_preferences AS target
USING (VALUES ($1::bigint, $2::text, $3::jsonb)) 
    AS source(user_id, pref_key, pref_value)
ON target.user_id = source.user_id 
   AND target.pref_key = source.pref_key
WHEN MATCHED THEN
    UPDATE SET
        pref_value = source.pref_value,
        updated_at = NOW()
WHEN NOT MATCHED THEN
    INSERT (user_id, pref_key, pref_value, created_at, updated_at)
    VALUES (source.user_id, source.pref_key, source.pref_value, NOW(), NOW())
RETURNING 
    target.id,
    target.created_at,
    CASE WHEN target.updated_at = target.created_at THEN 'inserted' ELSE 'updated' END AS action;

-- ETL-style bulk upsert with source staging table
MERGE INTO product_catalog AS target
USING staging_products AS source
ON target.sku = source.sku
WHEN MATCHED AND source.updated_at > target.updated_at THEN
    UPDATE SET
        name = source.name,
        price = source.price,
        inventory = source.inventory,
        updated_at = source.updated_at
WHEN MATCHED AND source.deleted = true THEN
    DELETE
WHEN NOT MATCHED AND source.deleted = false THEN
    INSERT (sku, name, price, inventory, created_at, updated_at)
    VALUES (source.sku, source.name, source.price, source.inventory, NOW(), source.updated_at)
RETURNING target.sku, merge_action() AS action;
-- merge_action() returns 'INSERT', 'UPDATE', or 'DELETE'

-- Create a logical replication slot that will survive failover
SELECT pg_create_logical_replication_slot(
    'my_subscription_slot',
    'pgoutput',
    false,  -- not temporary
    true    -- failover = true (new in PG 17)
);

-- Verify failover status
SELECT slot_name, failover, active
FROM pg_replication_slots
WHERE slot_name = 'my_subscription_slot';

-- On publisher: enable two-phase commit
CREATE PUBLICATION my_pub 
FOR TABLE orders, order_items
WITH (two_phase = true);

-- Subscriber
CREATE SUBSCRIPTION my_sub
CONNECTION 'host=primary port=5432 dbname=mydb'
PUBLICATION my_pub
WITH (two_phase = enable);

-- Sample: API response stored as JSON
SELECT *
FROM JSON_TABLE(
    '[
        {"id": 1, "name": "Alice", "scores": [95, 87, 92], "meta": {"dept": "eng"}},
        {"id": 2, "name": "Bob", "scores": [78, 83, 91], "meta": {"dept": "product"}},
        {"id": 3, "name": "Carol", "scores": [88, 94, 96], "meta": {"dept": "eng"}}
    ]'::json,
    '$[*]'
    COLUMNS (
        id       integer     PATH '$.id',
        name     text        PATH '$.name',
        dept     text        PATH '$.meta.dept',
        score_1  integer     PATH '$.scores[0]',
        score_2  integer     PATH '$.scores[1]',
        score_3  integer     PATH '$.scores[2]'
    )
) AS jt;

-- Result:
-- id | name  | dept    | score_1 | score_2 | score_3
-- ---+-------+---------+---------+---------+--------
--  1 | Alice | eng     |      95 |      87 |      92
--  2 | Bob   | product |      78 |      83 |      91
--  3 | Carol | eng     |      88 |      94 |      96

-- Parse structured JSON logs stored in a table
SELECT
    log_date,
    jt.request_id,
    jt.method,
    jt.path,
    jt.status_code,
    jt.duration_ms,
    jt.user_id
FROM application_logs,
JSON_TABLE(
    log_line,
    '$'
    COLUMNS (
        request_id  text    PATH '$.request_id',
        method      text    PATH '$.method',
        path        text    PATH '$.path',
        status_code integer PATH '$.status',
        duration_ms numeric PATH '$.duration_ms',
        user_id     bigint  PATH '$.user.id'
    )
) AS jt
WHERE log_date >= CURRENT_DATE - INTERVAL '7 days'
  AND jt.status_code >= 500
ORDER BY jt.duration_ms DESC
LIMIT 100;

-- Benchmark: 100M row aggregation, work_mem = 64MB
-- PG 16: ~42 seconds (heavy disk spill)
-- PG 17: ~28 seconds (better spill management)
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT 
    date_trunc('hour', created_at) AS hour,
    status,
    COUNT(*) AS event_count,
    SUM(value) AS total_value,
    AVG(value) AS avg_value,
    PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY value) AS p95_value
FROM large_events_table
WHERE created_at >= NOW() - INTERVAL '30 days'
GROUP BY 1, 2
ORDER BY 1 DESC, 2;

-- Monitor vacuum health
SELECT 
    schemaname,
    tablename,
    n_dead_tup,
    n_live_tup,
    ROUND(n_dead_tup::numeric / NULLIF(n_live_tup + n_dead_tup, 0) * 100, 2) AS dead_pct,
    last_vacuum,
    last_autovacuum,
    last_analyze,
    vacuum_count,
    autovacuum_count
FROM pg_stat_user_tables
WHERE n_dead_tup > 10000
ORDER BY n_dead_tup DESC
LIMIT 20;

-- Check if incremental backup is configured
SHOW summarize_wal;

-- List replication slots with failover status
SELECT slot_name, slot_type, failover, active, 
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS lag
FROM pg_replication_slots;

-- New PG 17: merge_action() in MERGE RETURNING
-- New PG 17: JSON_TABLE()
-- New PG 17: pg_basebackup --incremental

-- System stats: new in PG 17
SELECT * FROM pg_stat_io;   -- Detailed I/O stats per backend type
SELECT * FROM pg_stat_wal;  -- WAL activity stats