JSON_TABLE 일괄 처리

SET @bad_payload = JSON_OBJECT('posts', JSON_ARRAY(
  JSON_OBJECT('itemKey', 'a', 'requestKey', 'post-001', 'title', '정상 게시글',
    'content', '본문', 'status', 'PUBLISHED',
    'publishedAt', '2026-07-17 10:00:00'),
  JSON_OBJECT('itemKey', 'b', 'title', '요청 키 없는 게시글',
    'content', '본문', 'status', 'PUBLISHED', 'publishedAt', 'not-a-date')
));

SELECT x.* FROM JSON_TABLE(@bad_payload, '$.posts[*]' COLUMNS (
  item_key VARCHAR(80) PATH '$.itemKey',
  request_key VARCHAR(80) PATH '$.requestKey',
  title VARCHAR(200) PATH '$.title',
  content TEXT PATH '$.content',
  published_at DATETIME(6) PATH '$.publishedAt' ERROR ON ERROR
)) AS x;

DROP PROCEDURE IF EXISTS process_post_import;
DROP TABLE IF EXISTS post_import_rejection;
DROP TABLE IF EXISTS post_import_item;
DROP TABLE IF EXISTS post_import_batch;

CREATE TABLE post_import_batch (
  batch_id BINARY(16) PRIMARY KEY,
  author_id BIGINT UNSIGNED NOT NULL,
  payload JSON NOT NULL,
  payload_sha256 BINARY(32) NOT NULL,
  batch_status VARCHAR(16) NOT NULL DEFAULT 'STAGED',
  created_at DATETIME(6) NOT NULL,
  processed_at DATETIME(6) NULL,
  FOREIGN KEY (author_id) REFERENCES members (id) ON DELETE RESTRICT,
  CHECK (batch_status IN ('STAGED', 'REJECTED', 'COMPLETED'))
) ENGINE=InnoDB;
CREATE TABLE post_import_item (
  batch_id BINARY(16) NOT NULL,
  item_key VARCHAR(80) NOT NULL,
  post_id BIGINT UNSIGNED NOT NULL,
  PRIMARY KEY (batch_id, item_key),
  UNIQUE KEY uq_import_post (post_id),
  FOREIGN KEY (batch_id) REFERENCES post_import_batch (batch_id) ON DELETE RESTRICT,
  FOREIGN KEY (post_id) REFERENCES posts (id) ON DELETE RESTRICT
) ENGINE=InnoDB;
CREATE TABLE post_import_rejection (
  batch_id BINARY(16) NOT NULL,
  item_key VARCHAR(80) NOT NULL,
  error_code VARCHAR(40) NOT NULL,
  raw_item JSON NOT NULL,
  PRIMARY KEY (batch_id, item_key, error_code),
  FOREIGN KEY (batch_id) REFERENCES post_import_batch (batch_id) ON DELETE CASCADE
) ENGINE=InnoDB;

DELIMITER //
CREATE PROCEDURE process_post_import(IN p_batch BINARY(16))
proc: BEGIN
  DECLARE v_batch_status VARCHAR(16);
  DECLARE v_author BIGINT UNSIGNED;
  DECLARE v_payload JSON;
  DECLARE v_done BOOLEAN DEFAULT FALSE;
  DECLARE v_batch_found BOOLEAN DEFAULT TRUE;
  DECLARE v_key VARCHAR(80);
  DECLARE v_request_key VARCHAR(80);
  DECLARE v_title VARCHAR(200);
  DECLARE v_content LONGTEXT;
  DECLARE v_post_status VARCHAR(16);
  DECLARE v_published DATETIME(6);
  DECLARE v_post BIGINT UNSIGNED;
  DECLARE cur CURSOR FOR
    SELECT item_key, request_key, title, content, post_status, published_at
    FROM parsed_import ORDER BY ordinal_no;
  DECLARE CONTINUE HANDLER FOR NOT FOUND SET v_done = TRUE;
  DECLARE EXIT HANDLER FOR SQLEXCEPTION BEGIN ROLLBACK; RESIGNAL; END;

  START TRANSACTION;
  BEGIN
    DECLARE CONTINUE HANDLER FOR NOT FOUND SET v_batch_found = FALSE;
    SELECT batch_status, author_id, payload
      INTO v_batch_status, v_author, v_payload
    FROM post_import_batch
    WHERE batch_id = p_batch
    FOR UPDATE;
  END;
  IF v_batch_found = FALSE THEN
    SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'import batch not found';
  END IF;
  IF v_batch_status = 'COMPLETED' THEN
    COMMIT;
    SELECT item_key, post_id FROM post_import_item
    WHERE batch_id = p_batch ORDER BY item_key;
    LEAVE proc;
  END IF;

  DELETE FROM post_import_rejection WHERE batch_id = p_batch;
  DROP TEMPORARY TABLE IF EXISTS parsed_import;
  CREATE TEMPORARY TABLE parsed_import (
    ordinal_no INT UNSIGNED,
    item_key VARCHAR(80),
    request_key VARCHAR(80),
    title VARCHAR(200),
    content LONGTEXT,
    post_status VARCHAR(16),
    published_at DATETIME(6),
    raw_item JSON,
    PRIMARY KEY (ordinal_no)
  );
  INSERT INTO parsed_import
  SELECT j.ordinal_no, j.item_key, j.request_key, j.title, j.content,
         j.post_status, j.published_at, j.raw_item
  FROM JSON_TABLE(v_payload, '$.posts[*]' COLUMNS (
    ordinal_no FOR ORDINALITY,
    item_key VARCHAR(80) PATH '$.itemKey' NULL ON ERROR,
    request_key VARCHAR(80) PATH '$.requestKey' NULL ON ERROR,
    title VARCHAR(200) PATH '$.title' NULL ON ERROR,
    content TEXT PATH '$.content' NULL ON ERROR,
    post_status VARCHAR(16) PATH '$.status' NULL ON ERROR,
    published_at DATETIME(6) PATH '$.publishedAt' NULL ON EMPTY NULL ON ERROR,
    raw_item JSON PATH '$'
  )) AS j;

  INSERT INTO post_import_rejection
  SELECT p_batch, COALESCE(item_key, CONCAT('#', ordinal_no)),
         'MISSING_REQUIRED', raw_item
  FROM parsed_import
  WHERE item_key IS NULL OR request_key IS NULL OR title IS NULL
     OR content IS NULL OR post_status IS NULL;
  INSERT IGNORE INTO post_import_rejection
  SELECT p_batch, COALESCE(item_key, CONCAT('#', ordinal_no)),
         'INVALID_STATUS', raw_item
  FROM parsed_import
  WHERE post_status NOT IN ('DRAFT', 'PUBLISHED', 'HIDDEN');
  INSERT IGNORE INTO post_import_rejection
  SELECT p_batch, COALESCE(item_key, CONCAT('#', ordinal_no)),
         'INVALID_PUBLISH_TIME', raw_item
  FROM parsed_import
  WHERE (post_status = 'PUBLISHED') <> (published_at IS NOT NULL);
  INSERT IGNORE INTO post_import_rejection
  SELECT p_batch, COALESCE(p.item_key, CONCAT('#', p.ordinal_no)),
         'DUPLICATE_REQUEST_KEY', p.raw_item
  FROM (
    SELECT ordinal_no, item_key, request_key, raw_item,
           COUNT(*) OVER (PARTITION BY request_key) AS request_count
    FROM parsed_import
  ) AS p
  WHERE p.request_key IS NOT NULL AND p.request_count > 1;
  INSERT IGNORE INTO post_import_rejection
  SELECT p_batch, COALESCE(p.item_key, CONCAT('#', p.ordinal_no)),
         'ALREADY_EXISTS', p.raw_item
  FROM parsed_import AS p
  JOIN posts AS x
    ON x.author_id = v_author AND x.request_key = p.request_key;

  IF EXISTS (SELECT 1 FROM post_import_rejection WHERE batch_id = p_batch) THEN
    UPDATE post_import_batch
    SET batch_status = 'REJECTED', processed_at = CURRENT_TIMESTAMP(6)
    WHERE batch_id = p_batch;
    COMMIT;
    DROP TEMPORARY TABLE parsed_import;
    SELECT item_key, error_code FROM post_import_rejection
    WHERE batch_id = p_batch ORDER BY item_key, error_code;
    LEAVE proc;
  END IF;

  SET v_done = FALSE;
  OPEN cur;
  read_loop: LOOP
    FETCH cur INTO v_key, v_request_key, v_title, v_content,
                   v_post_status, v_published;
    IF v_done THEN LEAVE read_loop; END IF;
    INSERT INTO posts
      (author_id, title, content, status, view_count, created_at,
       updated_at, published_at, version, request_key)
    VALUES
      (v_author, v_title, v_content, v_post_status, 0, CURRENT_TIMESTAMP(6),
       CURRENT_TIMESTAMP(6), v_published, 0, v_request_key);
    SET v_post = LAST_INSERT_ID();
    INSERT INTO post_import_item VALUES (p_batch, v_key, v_post);
  END LOOP;
  CLOSE cur;
  UPDATE post_import_batch
  SET batch_status = 'COMPLETED', processed_at = CURRENT_TIMESTAMP(6)
  WHERE batch_id = p_batch;
  COMMIT;
  DROP TEMPORARY TABLE parsed_import;
  SELECT item_key, post_id FROM post_import_item
  WHERE batch_id = p_batch ORDER BY item_key;
END//
DELIMITER ;

INSERT INTO members (email, password_hash, name, status, created_at)
VALUES ('[email protected]', 'not-a-real-hash',
        'JSON 가져오기 예제 작성자', 'ACTIVE', CURRENT_TIMESTAMP(6))
ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id);
SET @import_author_id = LAST_INSERT_ID();

DELETE FROM posts
WHERE author_id = @import_author_id AND request_key LIKE 'ch18-2-%';

SET @completed_batch_id = UUID_TO_BIN('018f0000-0000-7000-8000-000000000201');
SET @completed_payload = JSON_OBJECT('posts', JSON_ARRAY(
  JSON_OBJECT(
    'itemKey', 'normal-1', 'requestKey', 'ch18-2-normal-1',
    'title', '게시판 사용 규칙', 'content', '회원 게시판의 사용 규칙을 정리합니다.',
    'status', 'PUBLISHED', 'publishedAt',
    DATE_FORMAT(CURRENT_TIMESTAMP(6) + INTERVAL 1 DAY, '%Y-%m-%d %H:%i:%s.%f')
  ),
  JSON_OBJECT(
    'itemKey', 'normal-2', 'requestKey', 'ch18-2-normal-2',
    'title', '다음 글 초안', 'content', '다음 게시글의 초안을 저장합니다.',
    'status', 'DRAFT'
  )
));
INSERT INTO post_import_batch
  (batch_id, author_id, payload, payload_sha256, batch_status, created_at)
VALUES (@completed_batch_id, @import_author_id, @completed_payload,
        UNHEX(SHA2(CAST(@completed_payload AS CHAR), 256)),
        'STAGED', CURRENT_TIMESTAMP(6));
CALL process_post_import(@completed_batch_id);

SET @posts_before_replay = (
  SELECT COUNT(*) FROM posts
  WHERE author_id = @import_author_id
    AND request_key IN ('ch18-2-normal-1', 'ch18-2-normal-2')
);
CALL process_post_import(@completed_batch_id);
SET @posts_after_replay = (
  SELECT COUNT(*) FROM posts
  WHERE author_id = @import_author_id
    AND request_key IN ('ch18-2-normal-1', 'ch18-2-normal-2')
);

SET @rejected_batch_id = UUID_TO_BIN('018f0000-0000-7000-8000-000000000202');
SET @rejected_payload = JSON_OBJECT('posts', JSON_ARRAY(
  JSON_OBJECT(
    'itemKey', 'retry-1', 'requestKey', 'ch18-2-retry-1',
    'title', '회원 가입 안내', 'content', '회원 가입 순서를 안내합니다.',
    'status', 'PUBLISHED', 'publishedAt',
    DATE_FORMAT(CURRENT_TIMESTAMP(6) + INTERVAL 1 DAY, '%Y-%m-%d %H:%i:%s.%f')
  ),
  JSON_OBJECT(
    'itemKey', 'retry-2',
    'title', '게시판 이용 안내', 'content', '게시판 이용 순서를 안내합니다.',
    'status', 'PUBLISHED'
  )
));
INSERT INTO post_import_batch
  (batch_id, author_id, payload, payload_sha256, batch_status, created_at)
VALUES (@rejected_batch_id, @import_author_id, @rejected_payload,
        UNHEX(SHA2(CAST(@rejected_payload AS CHAR), 256)),
        'STAGED', CURRENT_TIMESTAMP(6));
CALL process_post_import(@rejected_batch_id);

SET @retry_batch_id = UUID_TO_BIN('018f0000-0000-7000-8000-000000000203');
SET @retry_payload = JSON_SET(
  @rejected_payload,
  '$.posts[1].requestKey', 'ch18-2-retry-2',
  '$.posts[1].publishedAt',
  DATE_FORMAT(CURRENT_TIMESTAMP(6) + INTERVAL 1 DAY, '%Y-%m-%d %H:%i:%s.%f')
);
INSERT INTO post_import_batch
  (batch_id, author_id, payload, payload_sha256, batch_status, created_at)
VALUES (@retry_batch_id, @import_author_id, @retry_payload,
        UNHEX(SHA2(CAST(@retry_payload AS CHAR), 256)),
        'STAGED', CURRENT_TIMESTAMP(6));
CALL process_post_import(@retry_batch_id);

SELECT fixture, batch_status, input_items, item_mappings, rejections
FROM (
  SELECT 'normal' AS fixture, b.batch_status,
         JSON_LENGTH(b.payload, '$.posts') AS input_items,
         (SELECT COUNT(*) FROM post_import_item AS i
          WHERE i.batch_id = b.batch_id) AS item_mappings,
         (SELECT COUNT(*) FROM post_import_rejection AS r
          WHERE r.batch_id = b.batch_id) AS rejections
  FROM post_import_batch AS b WHERE b.batch_id = @completed_batch_id
  UNION ALL
  SELECT 'rejected', b.batch_status, JSON_LENGTH(b.payload, '$.posts'),
         (SELECT COUNT(*) FROM post_import_item AS i WHERE i.batch_id = b.batch_id),
         (SELECT COUNT(*) FROM post_import_rejection AS r WHERE r.batch_id = b.batch_id)
  FROM post_import_batch AS b WHERE b.batch_id = @rejected_batch_id
  UNION ALL
  SELECT 'retry', b.batch_status, JSON_LENGTH(b.payload, '$.posts'),
         (SELECT COUNT(*) FROM post_import_item AS i WHERE i.batch_id = b.batch_id),
         (SELECT COUNT(*) FROM post_import_rejection AS r WHERE r.batch_id = b.batch_id)
  FROM post_import_batch AS b WHERE b.batch_id = @retry_batch_id
) AS fixture_result
ORDER BY FIELD(fixture, 'normal', 'rejected', 'retry');

SELECT @posts_after_replay - @posts_before_replay AS replay_created_posts;

item_key | error_code
retry-2  | INVALID_PUBLISH_TIME
retry-2  | MISSING_REQUIRED

fixture  | batch_status | input_items | item_mappings | rejections
normal   | COMPLETED    | 2           | 2             | 0
rejected | REJECTED     | 2           | 0             | 2
retry    | COMPLETED    | 2           | 2             | 0

replay_created_posts
0

SELECT b.batch_id FROM post_import_batch AS b
WHERE b.batch_status = 'COMPLETED'
  AND (SELECT COUNT(*) FROM post_import_item AS i WHERE i.batch_id = b.batch_id)
      <> JSON_LENGTH(b.payload, '$.posts');

SELECT i.batch_id, i.item_key FROM post_import_item AS i
LEFT JOIN posts AS p ON p.id = i.post_id
WHERE p.id IS NULL;

SELECT b.batch_id FROM post_import_batch AS b
WHERE b.batch_status = 'REJECTED'
  AND EXISTS (SELECT 1 FROM post_import_item AS i WHERE i.batch_id = b.batch_id);

SELECT batch_id, item_key FROM post_import_item
GROUP BY batch_id, item_key HAVING COUNT(*) <> 1;

SELECT p.author_id, p.request_key FROM posts AS p
GROUP BY p.author_id, p.request_key HAVING COUNT(*) > 1;