数据归档

ADX_REPORT_BACKFILL.md

@@ -6,6 +6,7 @@
 
 - 主库热表继续保留最近 30 天数据
 - 后台任务开启后，会把 30 天前数据按月搬到 `adx_report`
+- 归档时会先按 `created_at` 选出一批候选 `id`，再按同一批 `id` 搬运和删除，避免 `LIMIT` 边界误删
 - 月历史表命名规则：
   - `tencent_ad_bid_events_history_YYYYMM`
   - `honor_ad_bid_events_history_YYYYMM`

src/main/java/com/adx/tencent/report/AdBidReportStore.java

@@ -81,12 +81,16 @@ public class AdBidReportStore {
             Timestamp monthStart = Timestamp.valueOf(month.atDay(1).atStartOfDay());
             Timestamp monthEnd = Timestamp.valueOf(month.plusMonths(1).atDay(1).atStartOfDay());
             for (int i = 0; i < archiveMaxBatchesPerRun; i++) {
-                int inserted = insertHistoryBatch(conn, media, month, monthStart, monthEnd, cutoff);
-                int deleted = deleteHotBatch(conn, media, monthStart, monthEnd, cutoff);
+                List<Long> candidateIds = selectArchiveCandidateIds(conn, media, monthStart, monthEnd, cutoff);
+                if (candidateIds.isEmpty()) {
+                    break;
+                }
+                insertHistoryBatch(conn, media, month, candidateIds);
+                int deleted = deleteHotBatch(conn, media, month, candidateIds);
                 if (deleted > 0) {
                     archived += deleted;
                 }
-                if (inserted == 0 && deleted == 0) {
+                if (deleted == 0) {
                     break;
                 }
             }
@@ -121,39 +125,54 @@ public class AdBidReportStore {
         }
     }
 
-    private int insertHistoryBatch(Connection conn,
-                                   MediaTable media,
-                                   YearMonth month,
-                                   Timestamp monthStart,
-                                   Timestamp monthEnd,
-                                   Timestamp cutoff) throws Exception {
-        String table = historyTable(media, month);
-        String sql = "INSERT IGNORE INTO `" + reportSchema + "`.`" + table + "` " +
-                "SELECT * FROM `" + mainSchema + "`.`" + media.hotTable + "` " +
+    private List<Long> selectArchiveCandidateIds(Connection conn,
+                                                 MediaTable media,
+                                                 Timestamp monthStart,
+                                                 Timestamp monthEnd,
+                                                 Timestamp cutoff) throws Exception {
+        String sql = "SELECT id FROM `" + mainSchema + "`.`" + media.hotTable + "` " +
                 "WHERE created_at >= ? AND created_at < ? AND created_at < ? " +
-                "ORDER BY created_at ASC LIMIT ?";
+                "ORDER BY created_at ASC, id ASC LIMIT ?";
+        List<Long> ids = new ArrayList<>(archiveBatchSize);
         try (PreparedStatement ps = conn.prepareStatement(sql)) {
             ps.setTimestamp(1, monthStart);
             ps.setTimestamp(2, monthEnd);
             ps.setTimestamp(3, cutoff);
             ps.setInt(4, archiveBatchSize);
-            return ps.executeUpdate();
+            try (ResultSet rs = ps.executeQuery()) {
+                while (rs.next()) {
+                    ids.add(rs.getLong(1));
+                }
+            }
+        }
+        return ids;
+    }
+
+    private void insertHistoryBatch(Connection conn,
+                                    MediaTable media,
+                                    YearMonth month,
+                                    List<Long> candidateIds) throws Exception {
+        String table = historyTable(media, month);
+        String sql = "INSERT IGNORE INTO `" + reportSchema + "`.`" + table + "` " +
+                "SELECT * FROM `" + mainSchema + "`.`" + media.hotTable + "` " +
+                "WHERE id IN (" + placeholders(candidateIds.size()) + ") " +
+                "ORDER BY created_at ASC, id ASC";
+        try (PreparedStatement ps = conn.prepareStatement(sql)) {
+            bindIds(ps, candidateIds, 1);
+            ps.executeUpdate();
         }
     }
 
     private int deleteHotBatch(Connection conn,
                                MediaTable media,
-                               Timestamp monthStart,
-                               Timestamp monthEnd,
-                               Timestamp cutoff) throws Exception {
-        String sql = "DELETE FROM `" + mainSchema + "`.`" + media.hotTable + "` " +
-                "WHERE created_at >= ? AND created_at < ? AND created_at < ? " +
-                "ORDER BY created_at ASC LIMIT ?";
+                               YearMonth month,
+                               List<Long> candidateIds) throws Exception {
+        String historyTable = historyTable(media, month);
+        String sql = "DELETE hot FROM `" + mainSchema + "`.`" + media.hotTable + "` hot " +
+                "INNER JOIN `" + reportSchema + "`.`" + historyTable + "` hist ON hist.id = hot.id " +
+                "WHERE hot.id IN (" + placeholders(candidateIds.size()) + ")";
         try (PreparedStatement ps = conn.prepareStatement(sql)) {
-            ps.setTimestamp(1, monthStart);
-            ps.setTimestamp(2, monthEnd);
-            ps.setTimestamp(3, cutoff);
-            ps.setInt(4, archiveBatchSize);
+            bindIds(ps, candidateIds, 1);
             return ps.executeUpdate();
         }
     }
@@ -239,6 +258,27 @@ public class AdBidReportStore {
         }
     }
 
+    private static void bindIds(PreparedStatement ps, List<Long> ids, int startIndex) throws Exception {
+        int index = startIndex;
+        for (Long id : ids) {
+            ps.setLong(index++, id);
+        }
+    }
+
+    private static String placeholders(int count) {
+        if (count <= 0) {
+            throw new IllegalArgumentException("placeholder count must be positive");
+        }
+        StringBuilder sb = new StringBuilder(count * 2);
+        for (int i = 0; i < count; i++) {
+            if (i > 0) {
+                sb.append(',');
+            }
+            sb.append('?');
+        }
+        return sb.toString();
+    }
+
     private static String safeIdentifier(String value) {
         if (value == null || !value.matches("[A-Za-z0-9_]+")) {
             throw new IllegalArgumentException("invalid sql identifier: " + value);