CREATE EXTENSION imv;

CREATE MATERIALIZED VIEW [ IF NOT EXISTS ] table_name
    [ (column_name [, ...] ) ]
    [ USING method ]
    [ TABLESPACE tablespace_name ]
    WITH ( refresh_mode = 'incremental', storage_parameter [= value] [, ... ] )
    AS query
    [ WITH [ NO ] DATA ]

-- 订单表
CREATE TABLE orders (
    order_id VARCHAR(20) PRIMARY KEY,
    user_id INT NOT NULL,
    order_time TIMESTAMP NOT NULL,
    total_amount DECIMAL(10,2) NOT NULL,
    product_list TEXT,
    shipping_address TEXT,
    order_status VARCHAR(20) DEFAULT 'created' -- created, paid, shipped, completed
);
 
-- 插入示例数据
INSERT INTO orders VALUES
('ORD10001', 101, '2023-06-01 10:00:00', 299.99, 'P1001,P1002', '北京市海淀区', 'created'),
('ORD10002', 102, '2023-06-01 11:30:00', 159.50, 'P1003', '上海市浦东新区', 'created'),
('ORD10003', 101, '2023-06-02 09:15:00', 89.99, 'P1004,P1005', '北京市海淀区', 'created'),
('ORD10004', 103, '2023-06-03 14:20:00', 450.00, 'P1006,P1007', '广州市天河区', 'created'),
('ORD10005', 102, '2023-06-03 16:45:00', 120.00, 'P1008', '上海市浦东新区', 'created'),
('ORD10006', 101, '2023-06-04 09:30:00', 75.50, 'P1009', '北京市海淀区', 'created');
 
-- 支付表
CREATE TABLE payments (
    payment_id VARCHAR(20) PRIMARY KEY,
    order_id VARCHAR(20) NOT NULL,
    payment_time TIMESTAMP NOT NULL,
    payment_amount DECIMAL(10,2) NOT NULL,
    payment_method VARCHAR(20), -- alipay, wechat, credit_card
    payment_status VARCHAR(20) -- success, failed, processing
);
 
-- 插入示例数据
INSERT INTO payments VALUES
('PAY20001', 'ORD10001', '2023-06-01 10:05:23', 299.99, 'alipay', 'success'),
('PAY20002', 'ORD10002', '2023-06-01 11:35:45', 159.50, 'wechat', 'success'),
('PAY20003', 'ORD10003', '2023-06-02 09:20:10', 89.99, 'credit_card', 'failed'),
('PAY20004', 'ORD10004', '2023-06-03 14:25:30', 450.00, 'alipay', 'success'),
('PAY20005', 'ORD10005', '2023-06-03 16:50:15', 120.00, 'wechat', 'failed');
 
-- 订单与支付关联的物化视图
CREATE MATERIALIZED VIEW order_payment_summary 
USING HYBRID
WITH (refresh_mode='incremental') 
AS
SELECT
    o.order_id,
    o.user_id,
    o.order_time,
    o.total_amount,
    o.order_status,
    p.payment_id,
    p.payment_time,
    p.payment_method,
    p.payment_status
FROM
    orders o
LEFT JOIN
    payments p ON o.order_id = p.order_id;
 
-- 结果表条数
SELECT count(1) FROM order_payment_summary;
 count
-------
     6
(1 row)
 
 
-- 按用户统计订单和支付情况
SELECT
    user_id,
    COUNT(*) AS total_orders,
    SUM(CASE WHEN payment_status = 'success' THEN 1 ELSE 0 END) AS paid_orders,
    SUM(CASE WHEN payment_status = 'failed' THEN 1 ELSE 0 END) AS failed_payments,
    SUM(CASE WHEN payment_status = 'processing' THEN 1 ELSE 0 END) AS processing_payments
FROM
    order_payment_summary
GROUP BY
    user_id;
 user_id | total_orders | paid_orders | failed_payments | processing_payments
---------+--------------+-------------+-----------------+---------------------
     101 |            3 |           1 |               1 |                   0
     102 |            2 |           1 |               1 |                   0
     103 |            1 |           1 |               0 |                   0
(3 rows)    
 

-- 更新 ORD10003 的支付状态为成功（之前是失败）
UPDATE payments SET payment_status = 'success', payment_time = '2023-06-02 10:15:00'
WHERE payment_id = 'PAY20003';
 
-- 刷新 order_payment_summary
REFRESH MATERIALIZED VIEW order_payment_summary;
 
-- 按用户统计订单和支付情况，我们可以看到 101 user 的 paid_orders 和 failed_payments 已经更新
SELECT
    user_id,
    COUNT(*) AS total_orders,
    SUM(CASE WHEN payment_status = 'success' THEN 1 ELSE 0 END) AS paid_orders,
    SUM(CASE WHEN payment_status = 'failed' THEN 1 ELSE 0 END) AS failed_payments,
    SUM(CASE WHEN payment_status = 'processing' THEN 1 ELSE 0 END) AS processing_payments
FROM
    order_payment_summary
GROUP BY
    user_id;
 user_id | total_orders | paid_orders | failed_payments | processing_payments
---------+--------------+-------------+-----------------+---------------------
     101 |            3 |           2 |               0 |                   0
     102 |            2 |           1 |               1 |                   0
     103 |            1 |           1 |               0 |                   0    
 
 
-- 查询刷新历史表，可以看到上面的增量刷新只修改了一条记录
SELECT 
    oid::regclass, refresh_action, statistics 
FROM imv.pg_stat_refresh_history 
WHERE oid = 'order_payment_summary'::regclass AND refresh_action = 'incremental';
          oid          | refresh_action |     statistics
-----------------------+----------------+---------------------
 order_payment_summary | incremental    | {"affected_row": 1}
(1 row)
 

ALTER MATERIALIZED VIEW [ IF EXISTS ] view_name AS query;

-- 给原来的物化视图增加一列： payment_status_category，用来替代原来的 payment_status 列
ALTER MATERIALIZED VIEW order_payment_summary AS
SELECT
    o.order_id,
    o.user_id,
    o.order_time,
    p.payment_id,
    p.payment_time,
    p.payment_method,
    -- p.payment_status,
    CASE
        WHEN p.payment_status = 'success' THEN 'paid'
        WHEN p.payment_status = 'failed' THEN 'payment_failed'
        ELSE 'awaiting_payment'
    END AS payment_status_category
FROM
    orders o
LEFT JOIN
    payments p ON o.order_id = p.order_id;
 
 
-- 通过 refresh history，能看到这次 alter 语句触发了一次全量刷新
SELECT
    oid::regclass, refresh_action, refresh_action_reason, statistics
FROM imv.pg_stat_refresh_history
WHERE oid = 'order_payment_summary'::regclass;
          oid          | refresh_action | refresh_action_reason |     statistics
-----------------------+----------------+-----------------------+---------------------
 order_payment_summary | full           | alter query           | {"affected_row": 6}
 

SELECT * FROM t WHERE ts >= NOW() - INTERVAL '1 hour';
 
-- 这种写法会报错
SELECT *，NOW() FROM t;

 
-- set 语句除了可以设置内置的 guc 变量外，还可以设定自定义变量，需要定义成 a.b 的形式
set protonbase.watermark = '2025-10-24 14:00:00';
 
-- 自定义变量可以通过 show 命令来查看
show protonbase.watermark;
 protonbase.watermark
----------------------
 2025-10-24 14:00:00
(1 row)
 
-- 也可以通过 current_setting 函数来获取
select current_setting('protonbase.watermark');
   current_setting
---------------------
 2025-10-24 14:00:00
(1 row)
 
-- 在 MV 定义中使用 current_setting
CREATE MATERIALIZED VIEW imv_with_setting 
WITH (refresh_mode='incremental') AS
SELECT * FROM t WHERE ts >= current_setting('protonbase.watermark')::timestamptz;
 
-- 在每次 REFRESH 前设置合适的值
set protonbase.watermark = '2025-10-24 14:10:00';
REFRESH MATERIALIZED VIEW imv_with_setting;
 
set protonbase.watermark = '2025-10-24 14:20:00';
REFRESH MATERIALIZED VIEW imv_with_setting;

   SELECT ...
   FROM T
   WHERE key IS NOT NULL
   GROUP BY key;

-- ts 为分钟级时间字段，通常随业务时间递增
SELECT SUM(a)
FROM T
GROUP BY ts, b;

SELECT SUM(a)
FROM T
GROUP BY b, ts;

SELECT 
    COUNT(DISTINCT CASE WHEN x = 1 THEN user_id ELSE NULL) AS uv1,
    COUNT(DISTINCT CASE WHEN x = 2 THEN user_id ELSE NULL) AS uv2,
    COUNT(DISTINCT CASE WHEN x = 3 THEN user_id ELSE NULL) AS uv3,
...

SELECT 
    COUNT(DISTINCT user_id) FILTER (WHERE x = 1) AS uv1,
    COUNT(DISTINCT user_id) FILTER (WHERE x = 2) AS uv2,
    COUNT(DISTINCT user_id) FILTER (WHERE x = 3) AS uv3,
...

S INNER JOIN T ON S.k1 = T.k2

S INNER JOIN ΔT ON S.k1 = ΔT.k2

S.k1::text = T.k2

UPPER(S.k1) = T.k2

S.k1_norm = T.k2_norm

SELECT a, b, c
FROM (
  SELECT
    a, b, c,
    ROW_NUMBER() OVER (PARTITION BY ... ORDER BY ...) AS rn
  FROM t
)
WHERE rn = 1;          -- Top-1
-- 或者：WHERE rn <= 10;  -- Top-10

CREATE TABLE t (a int, b int);
 
CREATE MATERIALIZED VIEW imv_join
WITH (refresh_mode='incremental') AS
SELECT t1.a, t2.b
FROM t AS t1, t AS t2
WHERE t1.a = t2.a;

NOTICE:  suggested index: CREATE INDEX ON t (b ASC)
NOTICE:  suggested index: CREATE INDEX ON t (a ASC)

SELECT COUNT(DISTINCT user_id)
FROM user_activities
WHERE pt = current_date
AND user_id not in (
    -- 统计过去30天出现过的所有用户
    SELECT DISTINCT user_id    
    FROM user_activities
    WHERE pt >= current_date - internal '30 day' AND pt < current_date AND pt)

-- 创建基础增量物化视图，按时间粒度聚合用户交易数据
CREATE MATERIALIZED VIEW user_transaction_base
WITH (refresh_mode = 'incremental') AS
SELECT 
  user_id,
  DATE_TRUNC('hour', transaction_time) as transaction_hour,
  DATE_TRUNC('day', transaction_time) as transaction_day,
  COUNT(*) as transaction_count,
  SUM(amount) as total_amount,
  AVG(amount) as avg_amount,
  MAX(amount) as max_amount,
  COUNT(DISTINCT merchant_id) as merchant_count
FROM transactions 
GROUP BY user_id, DATE_TRUNC('hour', transaction_time), DATE_TRUNC('day', transaction_time);
 
-- 创建高级物化视图，基于基础物化视图进行进一步聚合
CREATE MATERIALIZED VIEW user_risk_metrics
WITH (refresh_mode = 'incremental') AS
SELECT
  user_id,
  transaction_day,
  SUM(transaction_count) as daily_transaction_count,
  SUM(total_amount) as daily_total_amount,
  AVG(avg_amount) as daily_avg_amount,
  MAX(max_amount) as daily_max_amount,
  AVG(transaction_count) as hourly_avg_transactions,
  STDDEV(transaction_count) as transaction_count_stddev
FROM user_transaction_base
GROUP BY user_id, transaction_day;

-- 创建基础增量物化视图，按设备和时间粒度聚合传感器数据
CREATE MATERIALIZED VIEW device_sensor_base
WITH (refresh_mode = 'incremental') AS
SELECT 
  device_id,
  device_type,
  DATE_TRUNC('hour', reading_time) as reading_hour,
  DATE_TRUNC('day', reading_time) as reading_day,
  AVG(temperature) as avg_temperature,
  MAX(temperature) as max_temperature,
  MIN(temperature) as min_temperature,
  AVG(pressure) as avg_pressure,
  STDDEV(vibration) as vibration_stddev,
  COUNT(*) as reading_count,
  COUNT(CASE WHEN temperature > 80 THEN 1 END) as high_temp_count
FROM sensor_readings 
GROUP BY device_id, device_type, DATE_TRUNC('hour', reading_time), DATE_TRUNC('day', reading_time);
 
-- 创建高级物化视图，基于基础物化视图计算设备健康指标
CREATE MATERIALIZED VIEW device_health_metrics
WITH (refresh_mode = 'incremental') AS
SELECT
  device_id,
  device_type,
  reading_day,
  AVG(avg_temperature) as daily_avg_temperature,
  MAX(max_temperature) as daily_max_temperature,
  AVG(vibration_stddev) as daily_avg_vibration_stddev,
  SUM(high_temp_count) as daily_high_temp_count,
  COUNT(*) as hourly_readings_count,
  -- 计算设备异常率
  SUM(CASE WHEN high_temp_count > 0 THEN 1 ELSE 0 END) * 100.0 / COUNT(*) as high_temp_rate
FROM device_sensor_base
GROUP BY device_id, device_type, reading_day;

-- 创建基础增量物化视图，按服务和时间粒度聚合日志数据
CREATE MATERIALIZED VIEW log_base_metrics
WITH (refresh_mode = 'incremental') AS
SELECT 
  service_name,
  DATE_TRUNC('hour', log_time) as log_hour,
  DATE_TRUNC('day', log_time) as log_day,
  log_level,
  COUNT(*) as log_count,
  COUNT(CASE WHEN log_level = 'ERROR' THEN 1 END) as error_count,
  COUNT(CASE WHEN log_level = 'WARN' THEN 1 END) as warn_count,
  COUNT(CASE WHEN log_level = 'INFO' THEN 1 END) as info_count,
  COUNT(DISTINCT host_ip) as active_hosts,
  APPROX_COUNT_DISTINCT(user_id) as affected_users
FROM application_logs 
GROUP BY service_name, DATE_TRUNC('hour', log_time), DATE_TRUNC('day', log_time), log_level;
 
-- 创建高级物化视图，基于基础物化视图计算系统健康度指标
CREATE MATERIALIZED VIEW system_health_metrics
WITH (refresh_mode = 'incremental') AS
SELECT
  service_name,
  log_day,
  log_hour,
  SUM(log_count) as total_logs,
  SUM(error_count) as total_errors,
  SUM(warn_count) as total_warnings,
  -- 计算错误率
  CASE 
    WHEN SUM(log_count) > 0 THEN SUM(error_count) * 100.0 / SUM(log_count)
    ELSE 0 
  END as error_rate,
  -- 计算警告率
  CASE 
    WHEN SUM(log_count) > 0 THEN SUM(warn_count) * 100.0 / SUM(log_count)
    ELSE 0 
  END as warning_rate,
  MAX(active_hosts) as peak_active_hosts,
  SUM(affected_users) as total_affected_users
FROM log_base_metrics
GROUP BY service_name, log_day, log_hour;