Working with Data Migrations

long lastId = readCheckpoint();
while (true) {
    List<Row> batch = jdbc.query(
        "SELECT id, payload FROM src WHERE id > ? ORDER BY id LIMIT 1000", lastId);
    if (batch.isEmpty()) break;
    for (Row r : batch) {
        jdbc.update("INSERT INTO dst(id, payload) VALUES(?, ?) "
                  + "ON CONFLICT (id) DO UPDATE SET payload = EXCLUDED.payload",
                  r.id, transform(r.payload));
    }
    lastId = batch.get(batch.size() - 1).id;
    writeCheckpoint(lastId);
    rateLimiter.acquire(batch.size());
}