222_group_usage_daily_rollups.sql 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138
  1. -- /admin/groups 分组用量日汇总。
  2. -- 迁移创建结构与源表失效触发器,历史数据由后台聚合作业按持久水位回填。
  3. CREATE TABLE IF NOT EXISTS usage_group_daily_rollups (
  4. bucket_date DATE NOT NULL,
  5. group_id BIGINT NOT NULL,
  6. actual_cost DECIMAL(20, 10) NOT NULL DEFAULT 0,
  7. computed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  8. PRIMARY KEY (bucket_date, group_id)
  9. );
  10. COMMENT ON TABLE usage_group_daily_rollups IS '按北京时间自然日聚合的分组实际费用。';
  11. COMMENT ON COLUMN usage_group_daily_rollups.bucket_date IS '北京时间自然日。';
  12. CREATE TABLE IF NOT EXISTS usage_group_rollup_state (
  13. id SMALLINT PRIMARY KEY CHECK (id = 1),
  14. closed_before DATE NOT NULL DEFAULT DATE '1970-01-01',
  15. retained_from TIMESTAMPTZ NOT NULL DEFAULT TIMESTAMPTZ '1970-01-01 00:00:00+00',
  16. updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
  17. );
  18. COMMENT ON TABLE usage_group_rollup_state IS '分组日汇总的单行发布水位。';
  19. COMMENT ON COLUMN usage_group_rollup_state.closed_before IS '已完整发布日桶的北京时间日期排他上界。';
  20. INSERT INTO usage_group_rollup_state (id, closed_before, retained_from)
  21. VALUES (1, DATE '1970-01-01', TIMESTAMPTZ '1970-01-01 00:00:00+00')
  22. ON CONFLICT (id) DO NOTHING;
  23. -- 已发布范围的源记录发生变化时,必须在同一事务内后退发布水位。
  24. -- DELETE/UPDATE 使用行级触发器,以覆盖外键级联、分区表和直接分区写入。
  25. CREATE OR REPLACE FUNCTION invalidate_group_usage_rollup_state()
  26. RETURNS TRIGGER
  27. LANGUAGE plpgsql
  28. AS $$
  29. DECLARE
  30. affected_date DATE;
  31. published_before DATE;
  32. BEGIN
  33. IF TG_OP = 'DELETE' THEN
  34. affected_date := (OLD.created_at AT TIME ZONE 'Asia/Shanghai')::date;
  35. ELSE
  36. IF OLD.group_id IS NULL THEN
  37. affected_date := (NEW.created_at AT TIME ZONE 'Asia/Shanghai')::date;
  38. ELSIF NEW.group_id IS NULL THEN
  39. affected_date := (OLD.created_at AT TIME ZONE 'Asia/Shanghai')::date;
  40. ELSE
  41. affected_date := LEAST(
  42. (OLD.created_at AT TIME ZONE 'Asia/Shanghai')::date,
  43. (NEW.created_at AT TIME ZONE 'Asia/Shanghai')::date
  44. );
  45. END IF;
  46. END IF;
  47. -- 即使当前已发布水位尚未越过受影响日期,也必须先锁行。
  48. -- 否则并发关闭作业可能在本事务之后把水位推进,覆盖本次失效。
  49. SELECT closed_before
  50. INTO published_before
  51. FROM usage_group_rollup_state
  52. WHERE id = 1
  53. FOR UPDATE;
  54. IF published_before > affected_date THEN
  55. UPDATE usage_group_rollup_state
  56. SET closed_before = LEAST(closed_before, affected_date),
  57. updated_at = NOW()
  58. WHERE id = 1;
  59. END IF;
  60. IF TG_OP = 'DELETE' THEN
  61. RETURN OLD;
  62. END IF;
  63. RETURN NEW;
  64. END;
  65. $$;
  66. -- INSERT 是网关高频路径。transition table 让每个批量 INSERT 只锁一次状态行;
  67. -- KEY SHARE 在普通写入之间兼容,但会与关闭作业的 FOR UPDATE 串行化。
  68. CREATE OR REPLACE FUNCTION invalidate_group_usage_rollup_state_after_insert()
  69. RETURNS TRIGGER
  70. LANGUAGE plpgsql
  71. AS $$
  72. DECLARE
  73. affected_date DATE;
  74. published_before DATE;
  75. BEGIN
  76. SELECT MIN((created_at AT TIME ZONE 'Asia/Shanghai')::date)
  77. INTO affected_date
  78. FROM inserted_usage_logs
  79. WHERE group_id IS NOT NULL;
  80. IF affected_date IS NULL THEN
  81. RETURN NULL;
  82. END IF;
  83. SELECT closed_before
  84. INTO published_before
  85. FROM usage_group_rollup_state
  86. WHERE id = 1
  87. FOR KEY SHARE;
  88. IF published_before > affected_date THEN
  89. UPDATE usage_group_rollup_state
  90. SET closed_before = LEAST(closed_before, affected_date),
  91. updated_at = NOW()
  92. WHERE id = 1;
  93. END IF;
  94. RETURN NULL;
  95. END;
  96. $$;
  97. DROP TRIGGER IF EXISTS usage_logs_group_rollup_invalidate_insert ON usage_logs;
  98. CREATE TRIGGER usage_logs_group_rollup_invalidate_insert
  99. AFTER INSERT ON usage_logs
  100. REFERENCING NEW TABLE AS inserted_usage_logs
  101. FOR EACH STATEMENT
  102. EXECUTE FUNCTION invalidate_group_usage_rollup_state_after_insert();
  103. DROP TRIGGER IF EXISTS usage_logs_group_rollup_invalidate_delete ON usage_logs;
  104. CREATE TRIGGER usage_logs_group_rollup_invalidate_delete
  105. AFTER DELETE ON usage_logs
  106. FOR EACH ROW
  107. WHEN (OLD.group_id IS NOT NULL)
  108. EXECUTE FUNCTION invalidate_group_usage_rollup_state();
  109. DROP TRIGGER IF EXISTS usage_logs_group_rollup_invalidate_update ON usage_logs;
  110. CREATE TRIGGER usage_logs_group_rollup_invalidate_update
  111. AFTER UPDATE OF created_at, group_id, actual_cost ON usage_logs
  112. FOR EACH ROW
  113. WHEN (
  114. (
  115. OLD.created_at IS DISTINCT FROM NEW.created_at
  116. OR OLD.group_id IS DISTINCT FROM NEW.group_id
  117. OR OLD.actual_cost IS DISTINCT FROM NEW.actual_cost
  118. )
  119. AND (OLD.group_id IS NOT NULL OR NEW.group_id IS NOT NULL)
  120. )
  121. EXECUTE FUNCTION invalidate_group_usage_rollup_state();