DML 進階操作：Upsert、MERGE、RETURNING 與 COPY 批量匯入 | PostgreSQL

-- 觀察 Tuple 版本資訊
SELECT xmin, xmax, ctid, id, name
FROM users
WHERE id = 1;

-- 若 email 已存在，忽略本次插入
INSERT INTO users (email, name)
VALUES ('alice@example.com', 'Alice')
ON CONFLICT (email)
DO NOTHING;

-- 若 email 已存在，更新 name 與 updated_at
INSERT INTO users (email, name, updated_at)
VALUES ('alice@example.com', 'Alice Updated', NOW())
ON CONFLICT (email)
DO UPDATE SET
    name       = EXCLUDED.name,
    updated_at = EXCLUDED.updated_at
WHERE users.updated_at < EXCLUDED.updated_at;  -- 可加條件過濾

-- 建立複合唯一索引
CREATE UNIQUE INDEX ON product_prices (product_id, currency_code);

-- Upsert 需在 ON CONFLICT 中列出所有索引欄位
INSERT INTO product_prices (product_id, currency_code, price)
VALUES (42, 'USD', 99.99)
ON CONFLICT (product_id, currency_code)
DO UPDATE SET price = EXCLUDED.price;

-- 建立部分唯一索引
CREATE UNIQUE INDEX users_email_active_idx
    ON users (email) WHERE active = true;

-- ON CONFLICT 需指定相同的 WHERE 條件
INSERT INTO users (email, name, active)
VALUES ('dave@example.com', 'Dave', true)
ON CONFLICT (email) WHERE active = true
DO UPDATE SET name = EXCLUDED.name;

MERGE INTO products AS p
USING product_staging AS s
ON (p.sku = s.sku)
-- 已存在且有變動 → 更新
WHEN MATCHED AND (p.price <> s.price OR p.stock <> s.stock) THEN
    UPDATE SET
        price      = s.price,
        stock      = s.stock,
        updated_at = NOW()
-- 已存在且無變動 → 不做事
WHEN MATCHED THEN
    DO NOTHING
-- 不存在 → 新增
WHEN NOT MATCHED THEN
    INSERT (sku, price, stock, created_at)
    VALUES (s.sku, s.price, s.stock, NOW());

MERGE INTO products AS p
USING product_staging AS s ON (p.sku = s.sku)
WHEN MATCHED THEN
    UPDATE SET price = s.price
WHEN NOT MATCHED THEN
    INSERT (sku, price) VALUES (s.sku, s.price)
RETURNING merge_action(), p.sku, p.price;
-- merge_action() 回傳 'INSERT' 或 'UPDATE'

-- INSERT RETURNING：取得自動生成的 id
INSERT INTO users (email, name)
VALUES ('charlie@example.com', 'Charlie')
RETURNING id, created_at;

-- UPDATE RETURNING：取得更新後的值
UPDATE products
SET price = price * 1.1
WHERE category = 'electronics'
RETURNING id, name, price AS new_price;

-- DELETE RETURNING：取得被刪除的資料
DELETE FROM sessions
WHERE expires_at < NOW()
RETURNING user_id, token, expires_at;

-- 將過期訊息歸檔 + 更新使用者計數，一條 SQL 完成
WITH
-- 步驟 1：刪除過期訊息
expired AS (
    DELETE FROM messages
    WHERE created_at < NOW() - INTERVAL '90 days'
    RETURNING id, user_id, content, created_at
),
-- 步驟 2：歸檔
archived AS (
    INSERT INTO message_archive (id, user_id, content, created_at, archived_at)
    SELECT id, user_id, content, created_at, NOW()
    FROM expired
    RETURNING id
),
-- 步驟 3：更新計數
updated_counts AS (
    UPDATE user_stats us
    SET message_count = message_count - sub.cnt
    FROM (
        SELECT user_id, COUNT(*) AS cnt
        FROM expired
        GROUP BY user_id
    ) sub
    WHERE us.user_id = sub.user_id
    RETURNING us.user_id, us.message_count
)
-- 回傳統計
SELECT
    (SELECT COUNT(*) FROM archived)       AS archived_messages,
    (SELECT COUNT(*) FROM updated_counts) AS updated_users;

-- 扣款並記錄交易流水
WITH deducted AS (
    UPDATE accounts
    SET balance = balance - 500
    WHERE user_id = 1 AND balance >= 500
    RETURNING user_id, balance AS new_balance
)
INSERT INTO transaction_log (user_id, amount, type, balance_after, created_at)
SELECT user_id, 500, 'debit', new_balance, NOW()
FROM deducted;
-- 若 deducted 為空（餘額不足），INSERT 也不執行

-- 從 CSV 匯入（伺服器端路徑）
COPY users (id, email, name, created_at)
FROM '/var/data/users.csv'
WITH (FORMAT csv, HEADER true, DELIMITER ',', NULL '');

-- 匯出至 CSV
COPY (SELECT * FROM users WHERE active = true)
TO '/var/data/active_users.csv'
WITH (FORMAT csv, HEADER true);

# psql 中執行
\copy users (id, email, name) FROM '/Users/me/users.csv' CSV HEADER
\copy (SELECT * FROM reports) TO '/Users/me/report.csv' CSV HEADER

-- 多行 VALUES 批量插入（建議每批 500-1000 行）
INSERT INTO events (user_id, event_type, payload)
VALUES
    (1, 'click', '{"btn": "submit"}'),
    (2, 'view',  '{"page": "/home"}'),
    (3, 'click', '{"btn": "cancel"}');

-- unnest 批量插入（適合程式傳入陣列）
INSERT INTO events (user_id, event_type)
SELECT * FROM unnest(
    ARRAY[1, 2, 3]::int[],
    ARRAY['click', 'view', 'click']::text[]
) AS t(user_id, event_type);

-- 每個分類取最貴的 3 個商品
SELECT c.name AS category, p.name AS product, p.price
FROM categories c
JOIN LATERAL (
    SELECT name, price
    FROM products
    WHERE category_id = c.id
    ORDER BY price DESC
    LIMIT 3
) p ON true
ORDER BY c.name, p.price DESC;

-- 展開 jsonb 陣列的每個元素
SELECT o.id AS order_id,
       item->>'name' AS item_name,
       (item->>'qty')::int AS qty
FROM orders o,
     LATERAL jsonb_array_elements(o.items) AS item;

-- 每個使用者：最近一次登入 + 登入後的第一個事件
SELECT u.id, u.email, last_login.logged_at, next_event.event_type
FROM users u
JOIN LATERAL (
    SELECT logged_at
    FROM login_history
    WHERE user_id = u.id
    ORDER BY logged_at DESC
    LIMIT 1
) last_login ON true
LEFT JOIN LATERAL (
    SELECT event_type, created_at
    FROM user_events
    WHERE user_id = u.id
      AND created_at > last_login.logged_at
    ORDER BY created_at ASC
    LIMIT 1
) next_event ON true;

-- TRUNCATE 並重置自增 ID
TRUNCATE TABLE staging_data RESTART IDENTITY;

-- 危險！CASCADE 會自動 TRUNCATE 外鍵連結的子表
TRUNCATE TABLE orders CASCADE;

-- 較安全：明確列出所有目標表
TRUNCATE TABLE order_items, orders;

-- 1. 停用 autovacuum（避免匯入中途觸發）
ALTER TABLE big_table SET (autovacuum_enabled = false);

-- 2. 刪除非必要索引（匯入後重建更快）
DROP INDEX IF EXISTS big_table_search_idx;

-- 3. 暫時停用觸發器
ALTER TABLE big_table DISABLE TRIGGER ALL;

-- 4. 設定 UNLOGGED（可接受 crash risk 時）
ALTER TABLE big_table SET UNLOGGED;

-- 5. 執行 COPY
COPY big_table FROM '/data/big_data.csv' CSV HEADER;

-- 6. 恢復設定
ALTER TABLE big_table SET LOGGED;
ALTER TABLE big_table ENABLE TRIGGER ALL;
CREATE INDEX big_table_search_idx ON big_table (search_col);
ALTER TABLE big_table SET (autovacuum_enabled = true);

-- 7. 更新統計
ANALYZE big_table;

-- 關閉同步提交提升吞吐量（非關鍵寫入可接受）
SET synchronous_commit = off;
-- 批量操作...
RESET synchronous_commit;

-- 提升 maintenance_work_mem 加速索引重建
SET maintenance_work_mem = '1GB';
CREATE INDEX big_table_col_idx ON big_table (col);
RESET maintenance_work_mem;

CREATE TEMP TABLE staging_import (LIKE target_table);
COPY staging_import FROM '/data/input.csv' CSV HEADER;

INSERT INTO target_table
SELECT * FROM staging_import
ON CONFLICT DO NOTHING;