-- create_date index
SELECT p.* FROM product p JOIN 
(SELECT id FROM product WHERE create_date = :date ORDER BY id ASC LIMIT :page, :pageSize) as temp 
on temp.id = p.id

// JdbcTemplate Paging Reader
public class CustomJdbcPagingItemReader<T> implements ItemReader<T> {

    private final NamedParameterJdbcTemplate jdbcTemplate;
    private final String query;
    private final RowMapper<T> rowMapper;
    private int page;
    private final int pageSize;
    private int currentItemIndex;
    private List<T> items;
    private LocalDate date;
    protected Log logger = LogFactory.getLog(getClass());

    public CustomJdbcPagingItemReader(DataSource dataSource, String query, RowMapper<T> rowMapper, int pageSize) {
        this.jdbcTemplate = new NamedParameterJdbcTemplate(dataSource);
        this.query = query;
        this.rowMapper = rowMapper;
        this.page = 0;
        this.pageSize = pageSize;
        this.currentItemIndex = 0;
    }

    public void setDate(LocalDate date) {
        this.date = date;
    }

    @Override
    public T read() {
        if (items == null || currentItemIndex >= items.size()) {
            fetchNextPage();
        }

        if (items == null || items.isEmpty()) {
            return null; // 더 이상 읽을 데이터가 없음
        }

        return items.get(currentItemIndex++);
    }

    private void fetchNextPage() {
        Map<String, Object> parameters = new HashMap<>();
        parameters.put("page", page * pageSize);
        parameters.put("pageSize", pageSize);
        parameters.put("date", date);
        items = jdbcTemplate.query(query, parameters, rowMapper);
        currentItemIndex = 0;
        page++;
    }
}

// Batch 작업
@JobScope
@Bean
public CustomJdbcPagingItemReader<Product> reader() {
    String query = "SELECT p.* FROM product p JOIN (SELECT id FROM product WHERE create_date = :date ORDER BY id ASC LIMIT :page, :pageSize) as temp on temp.id = p.id";
    CustomJdbcPagingItemReader<Product> reader = new CustomJdbcPagingItemReader<>(dataSource, query, new ProductRowMapper(), chunkSize);
    reader.setDate(jobParameter.getDate());

    return reader;
}

SELECT * FROM product WHERE create_date = ? ORDER BY id ASC limit 0, 100 -- 빠름

SELECT * FROM product WHERE create_date = ? ORDER BY id ASC limit 5000000, 100 -- OFFSET이 커질수록 느려짐

SELECT * FROM product WHERE create_date = ? AND id > 0 ORDER BY id ASC limit 0, 1000

SELECT * FROM product WHERE create_date = ? AND id > 5000001 ORDER BY id ASC limit 0, 1000

@JobScope
@Bean
public JdbcCursorItemReader<Product> reader() {
    return new JdbcCursorItemReaderBuilder<Product>()
        .name("jdbcCursorItemReader")
        .fetchSize(chunkSize)
        .sql("SELECT * FROM product WHERE create_date = ? ORDER BY id ASC")
        .rowMapper(new ProductRowMapper())
        .dataSource(dataSource)
        .preparedStatementSetter(ps -> ps.setDate(1, Date.valueOf(jobParameter.getDate())))
        .build();
}

@Bean
public ItemWriter<Product> redisItemWriter() {
    return products -> stringRedisTemplate.executePipelined((RedisCallback<Object>) connection -> {
        for (Product product : products) {
            String key = redisKeyPrefix + product.getCreateDate().toString();
            connection.stringCommands().incrBy(key.getBytes(), product.getAmount());
        }
        return null;
    });
}

@Bean
public StepExecutionListener redisToDatabaseSaver() {
    return new StepExecutionListener() {
        @Override
        public ExitStatus afterStep(@NonNull StepExecution stepExecution) {
            EntityManager entityManager = entityManagerFactory.createEntityManager();
            EntityTransaction transaction = entityManager.getTransaction();
            transaction.begin();

            try {
                Set<String> keys = redisTemplate.keys(redisKeyPrefix + "*");
                ValueOperations<String, Long> valueOps = redisTemplate.opsForValue();

                for (String key : keys) {
                    Long totalAmount = valueOps.get(key);
                    LocalDate createDate = LocalDate.parse(key.split(":")[1]);

                    ProductBackup backup = ProductBackup.builder()
                            .amount(totalAmount)
                            .createDate(createDate)
                            .build();
                    entityManager.merge(backup);
                }
                transaction.commit();
            } catch (Exception e) {
                transaction.rollback();
                throw e;
            } finally {
                entityManager.close();
            }
            return ExitStatus.COMPLETED;
        }
    };
}

@Bean
public JdbcBatchItemWriter<ProductBackup> writer() {
    JdbcBatchItemWriter<ProductBackup> writer = new JdbcBatchItemWriter<>();

    writer.setDataSource(dataSource);
    writer.setSql("INSERT INTO product_backup (name, amount, create_date) VALUES (:name, :amount, :createDate)");

    writer.setItemSqlParameterSourceProvider(
            item -> new MapSqlParameterSource()
                    .addValue("name", item.getName())
                    .addValue("amount", item.getAmount())
                    .addValue("createDate", item.getCreateDate())
    );

    return writer;
}

-- Chunk Size 1000 기준

-- 단일 UPDATE
-- Chunk Size만큼 Database I/O가 발생 (최대 1000번) 
UPDATE user SET grade = 'BASIC' WHERE id = 1;
-- ...
UPDATE user SET grade = 'NORMAL' WHERE id = 10;
-- ...
UPDATE user SET grade = 'VIP' WHERE id = 1000;

-- IN UPDATE
-- Chunk Size와 상관없이 최대 3번의 Database I/O가 발생 (최대 3번)
UPDATE user SET grade = 'BASIC' WHERE id in (1, 2, ...);
UPDATE user SET grade = 'NORMAL' WHERE id in (10 11, ...);
UPDATE user SET grade = 'VIP' WHERE id in (500, ...);

// Item Writer
public class CustomJdbcBatchItemWriter<T> implements ItemWriter<T> {
    private DataSource dataSource;
    private String sql;
    private BiConsumer<PreparedStatement, T> preparedStatementSetter;

    public CustomJdbcBatchItemWriter(DataSource dataSource, String sql,
                                     BiConsumer<PreparedStatement, T> preparedStatementSetter) {
        this.dataSource = dataSource;
        this.sql = sql;
        this.preparedStatementSetter = preparedStatementSetter;
    }

    @Override
    public void write(Chunk<? extends T> chunk) throws Exception {
        List<? extends T> items = chunk.getItems();
        Connection connection = null;
        PreparedStatement preparedStatement = null;

        try {
            connection = dataSource.getConnection(); // 커넥션 획득
            connection.setAutoCommit(false); // 자동 커밋 비활성화
            preparedStatement = connection.prepareStatement(this.sql); //sql 로 prepared statement 생성

            for (T item : items) {
                preparedStatementSetter.accept(preparedStatement, item); // prepared statement에 파라미터 바인딩
                preparedStatement.addBatch(); // batch에 prepared statement 추가
            }

            preparedStatement.executeBatch(); // 배치에 추가된 모든 SQL 문을 데이터베이스에 한 번에 전송
            connection.commit(); // 트랜잭션을 커밋
        } catch (SQLException e) {
            if (connection != null) {
                connection.rollback();
            }
            throw new Exception("Error executing batch write", e);
        } finally {
            if (preparedStatement != null) {
                preparedStatement.close();
            }
            if (connection != null) {
                connection.close();
            }
        }
    }
}

// Batch Writer
@Bean
public CustomJdbcBatchItemWriter<ProductBackup> writer() {
    String sql = "INSERT INTO product_backup (name, amount, create_date) VALUES (?, ?, ?)";

    return new CustomJdbcBatchItemWriter<>(dataSource, sql, (PreparedStatement ps, ProductBackup item) -> {
        try {
            ps.setString(1, item.getName());
            ps.setLong(2, item.getAmount());
            ps.setDate(3, Date.valueOf(item.getCreateDate()));
        } catch (SQLException e) {
            throw new RuntimeException("Error setting PreparedStatement values", e);
        }
    });
}