마이크로배치 집계

DROP TABLE IF EXISTS final_stat_delta_bad;
CREATE TABLE final_stat_delta_bad (
  author_id BIGINT UNSIGNED,
  stat_date DATE,
  total_views BIGINT UNSIGNED,
  PRIMARY KEY (author_id, stat_date)
) ENGINE=InnoDB;

INSERT INTO final_stat_delta_bad VALUES (1, '2026-07-15', 60);
-- 이벤트 101을 처리한 뒤 응답을 잃어 같은 내용을 다시 소비한다.
UPDATE final_stat_delta_bad SET total_views = total_views + 60
WHERE author_id = 1 AND stat_date = '2026-07-15';
-- 실제 게시글은 조회수 90으로 정정되었지만 소비자는 90을 다시 더한다.
UPDATE final_stat_delta_bad SET total_views = total_views + 90
WHERE author_id = 1 AND stat_date = '2026-07-15';

SELECT * FROM final_stat_delta_bad;

author_id | stat_date  | total_views
1         | 2026-07-15 | 210

DROP PROCEDURE IF EXISTS process_post_stats_microbatch;
DROP TABLE IF EXISTS daily_post_stats_incremental;
DROP TABLE IF EXISTS post_stat_contribution;
DROP TABLE IF EXISTS final_stat_checkpoint;
DROP TABLE IF EXISTS post_change_event;

CREATE TABLE post_change_event (
  event_id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
  post_id BIGINT UNSIGNED NOT NULL,
  source_version INT UNSIGNED NOT NULL,
  author_id BIGINT UNSIGNED NOT NULL,
  stat_date DATE NOT NULL,
  published_count TINYINT UNSIGNED NOT NULL,
  total_views BIGINT UNSIGNED NOT NULL,
  produced_at DATETIME(6) NOT NULL,
  CONSTRAINT uq_stat_event_version UNIQUE (post_id, source_version),
  CONSTRAINT chk_stat_event_value CHECK (
    (published_count = 0 AND total_views = 0)
    OR (published_count = 1 AND total_views <= 1000000000)
  ),
  INDEX ix_stat_event_stream (event_id)
) ENGINE=InnoDB;

CREATE TABLE post_stat_contribution (
  post_id BIGINT UNSIGNED PRIMARY KEY,
  source_version INT UNSIGNED NOT NULL,
  author_id BIGINT UNSIGNED NOT NULL,
  stat_date DATE NOT NULL,
  published_count TINYINT UNSIGNED NOT NULL,
  total_views BIGINT UNSIGNED NOT NULL,
  last_event_id BIGINT UNSIGNED NOT NULL,
  INDEX ix_contribution_day (author_id, stat_date)
) ENGINE=InnoDB;

CREATE TABLE daily_post_stats_incremental (
  author_id BIGINT UNSIGNED NOT NULL,
  stat_date DATE NOT NULL,
  published_count INT UNSIGNED NOT NULL,
  total_views BIGINT UNSIGNED NOT NULL,
  rebuilt_from_event BIGINT UNSIGNED NOT NULL,
  PRIMARY KEY (author_id, stat_date)
) ENGINE=InnoDB;

CREATE TABLE final_stat_checkpoint (
  consumer_name VARCHAR(80) PRIMARY KEY,
  last_event_id BIGINT UNSIGNED NOT NULL,
  updated_at DATETIME(6) NOT NULL
) ENGINE=InnoDB;
INSERT INTO final_stat_checkpoint VALUES ('daily-stat-v1', 0, '2026-07-16 00:00:00');

DELIMITER //
CREATE PROCEDURE process_post_stats_microbatch(
  IN p_consumer VARCHAR(80),
  IN p_limit INT UNSIGNED
)
proc: BEGIN
  DECLARE v_from BIGINT UNSIGNED;
  DECLARE v_to BIGINT UNSIGNED;
  DECLARE EXIT HANDLER FOR SQLEXCEPTION
  BEGIN
    ROLLBACK;
    DROP TEMPORARY TABLE IF EXISTS next_contribution;
    DROP TEMPORARY TABLE IF EXISTS affected_stat_key;
    RESIGNAL;
  END;

  IF p_consumer IS NULL OR p_limit IS NULL OR p_limit = 0 THEN
    SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'invalid consumer or batch limit';
  END IF;

  START TRANSACTION;
  SELECT last_event_id INTO v_from
  FROM final_stat_checkpoint
  WHERE consumer_name = p_consumer
  FOR UPDATE;
  IF v_from IS NULL THEN
    SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'consumer checkpoint not registered';
  END IF;

  SELECT MAX(event_id) INTO v_to
  FROM (
    SELECT event_id FROM post_change_event
    WHERE event_id > v_from
    ORDER BY event_id
    LIMIT p_limit
  ) AS batch_ids;
  IF v_to IS NULL THEN
    COMMIT;
    SELECT v_from AS checkpoint, 'EMPTY' AS outcome;
    LEAVE proc;
  END IF;

  CREATE TEMPORARY TABLE next_contribution (
    post_id BIGINT UNSIGNED PRIMARY KEY,
    source_version INT UNSIGNED,
    author_id BIGINT UNSIGNED,
    stat_date DATE,
    published_count TINYINT UNSIGNED,
    total_views BIGINT UNSIGNED,
    event_id BIGINT UNSIGNED
  ) ENGINE=InnoDB;
  INSERT INTO next_contribution
    (post_id, source_version, author_id, stat_date,
     published_count, total_views, event_id)
  SELECT post_id, source_version, author_id, stat_date,
         published_count, total_views, event_id
  FROM (
    SELECT e.*,
           ROW_NUMBER() OVER (
             PARTITION BY e.post_id
             ORDER BY e.source_version DESC, e.event_id DESC
           ) AS version_rank
    FROM post_change_event AS e
    WHERE e.event_id > v_from AND e.event_id <= v_to
  ) AS ranked
  WHERE version_rank = 1;

  CREATE TEMPORARY TABLE affected_stat_key (
    author_id BIGINT UNSIGNED,
    stat_date DATE,
    PRIMARY KEY (author_id, stat_date)
  ) ENGINE=InnoDB;
  INSERT IGNORE INTO affected_stat_key
  SELECT c.author_id, c.stat_date
  FROM post_stat_contribution AS c
  JOIN next_contribution AS n ON n.post_id = c.post_id
  WHERE n.source_version > c.source_version;
  INSERT IGNORE INTO affected_stat_key
  SELECT n.author_id, n.stat_date
  FROM next_contribution AS n
  LEFT JOIN post_stat_contribution AS c ON c.post_id = n.post_id
  WHERE c.post_id IS NULL OR n.source_version > c.source_version;

  UPDATE post_stat_contribution AS c
  JOIN next_contribution AS n ON n.post_id = c.post_id
  SET c.source_version = n.source_version,
      c.author_id = n.author_id,
      c.stat_date = n.stat_date,
      c.published_count = n.published_count,
      c.total_views = n.total_views,
      c.last_event_id = n.event_id
  WHERE n.source_version > c.source_version;
  INSERT INTO post_stat_contribution
    (post_id, source_version, author_id, stat_date,
     published_count, total_views, last_event_id)
  SELECT n.post_id, n.source_version, n.author_id, n.stat_date,
         n.published_count, n.total_views, n.event_id
  FROM next_contribution AS n
  LEFT JOIN post_stat_contribution AS c ON c.post_id = n.post_id
  WHERE c.post_id IS NULL;

  DELETE d FROM daily_post_stats_incremental AS d
  JOIN affected_stat_key AS a
    ON a.author_id = d.author_id AND a.stat_date = d.stat_date;
  INSERT INTO daily_post_stats_incremental
  SELECT c.author_id, c.stat_date, SUM(c.published_count), SUM(c.total_views), v_to
  FROM post_stat_contribution AS c
  JOIN affected_stat_key AS a
    ON a.author_id = c.author_id AND a.stat_date = c.stat_date
  GROUP BY c.author_id, c.stat_date
  HAVING SUM(c.published_count) > 0;

  UPDATE final_stat_checkpoint
  SET last_event_id = v_to, updated_at = CURRENT_TIMESTAMP(6)
  WHERE consumer_name = p_consumer;
  COMMIT;
  DROP TEMPORARY TABLE next_contribution;
  DROP TEMPORARY TABLE affected_stat_key;
  SELECT v_to AS checkpoint, 'APPLIED' AS outcome;
END//
DELIMITER ;

SET @post_id = (
  SELECT id FROM posts WHERE status = 'PUBLISHED' ORDER BY id LIMIT 1
);
SET @author_id = (SELECT author_id FROM posts WHERE id = @post_id);
INSERT INTO post_change_event
  (post_id, source_version, author_id, stat_date,
   published_count, total_views, produced_at)
VALUES (@post_id, 1, @author_id, '2026-07-15', 1, 60, '2026-07-16 01:00:00');

CALL process_post_stats_microbatch('daily-stat-v1', 100);
CALL process_post_stats_microbatch('daily-stat-v1', 100);

SELECT post_id, source_version, stat_date, total_views, last_event_id
FROM post_stat_contribution;
SELECT author_id, stat_date, published_count, total_views, rebuilt_from_event
FROM daily_post_stats_incremental;

checkpoint | outcome
1          | APPLIED

checkpoint | outcome
1          | EMPTY

post_id | source_version | stat_date  | total_views | last_event_id
1       | 1              | 2026-07-15 | 60          | 1

author_id | stat_date  | published_count | total_views | rebuilt_from_event
1         | 2026-07-15 | 1               | 60          | 1

INSERT INTO post_change_event
  (post_id, source_version, author_id, stat_date,
   published_count, total_views, produced_at)
VALUES
  (@post_id, 3, @author_id, '2026-07-16', 1, 90, '2026-07-16 03:00:00'),
  (@post_id, 2, @author_id, '2026-07-15', 1, 90, '2026-07-16 02:00:00');
CALL process_post_stats_microbatch('daily-stat-v1', 100);
CALL process_post_stats_microbatch('daily-stat-v1', 100);

SELECT author_id, stat_date, published_count, total_views, rebuilt_from_event
FROM daily_post_stats_incremental;

WITH expected AS (
  SELECT author_id, stat_date,
         SUM(published_count) AS published_count,
         SUM(total_views) AS total_views
  FROM post_stat_contribution
  GROUP BY author_id, stat_date
  HAVING SUM(published_count) > 0
)
SELECT e.author_id, e.stat_date
FROM expected AS e
LEFT JOIN daily_post_stats_incremental AS d
  ON d.author_id = e.author_id AND d.stat_date = e.stat_date
WHERE d.author_id IS NULL
   OR d.published_count <> e.published_count
   OR d.total_views <> e.total_views
UNION ALL
SELECT d.author_id, d.stat_date
FROM daily_post_stats_incremental AS d
LEFT JOIN expected AS e
  ON e.author_id = d.author_id AND e.stat_date = d.stat_date
WHERE e.author_id IS NULL;

SELECT e.post_id, MAX(e.source_version) AS expected_version,
       c.source_version AS actual_version
FROM post_change_event AS e
JOIN post_stat_contribution AS c ON c.post_id = e.post_id
GROUP BY e.post_id, c.source_version
HAVING c.source_version <> MAX(e.source_version);

SELECT earlier.event_id AS higher_version_event,
       later.event_id AS late_lower_version_event
FROM post_change_event AS earlier
JOIN post_change_event AS later
  ON later.post_id = earlier.post_id AND later.event_id > earlier.event_id
WHERE later.source_version < earlier.source_version;

checkpoint | outcome
3          | APPLIED

checkpoint | outcome
3          | EMPTY

author_id | stat_date  | published_count | total_views | rebuilt_from_event
1         | 2026-07-16 | 1               | 90          | 3

higher_version_event | late_lower_version_event
2                    | 3