Batch Insert 성능 향상기 2편 - 성능 측정

const val GLOBAL_CHUNK_SIZE = 1_000
const val DATA_SET_UP_SIZE = 5_000_000

@Configuration
class BulkInsertJobConfiguration(
    private val jobBuilderFactory: JobBuilderFactory,
    private val jobDataSetUpListener: JobDataSetUpListener,
    private val dataSource: DataSource,
    private val exposedDataBase: Database,
    private val paymentBackJpaRepository: PaymentBackJpaRepository,
    entityManagerFactory: EntityManagerFactory
) {

    @Bean
    fun bulkInsertJob(
        jdbcTemplateInsertStep: Step
    ): Job =
        jobBuilderFactory["bulkInsertJob"]
            .incrementer(RunIdIncrementer())
            .listener(jobDataSetUpListener) // payment 데이터 setup
            .start(jdbcTemplateInsertStep)
            .build()

    @Bean
    @JobScope
    fun bulkInsertStep(
        stepBuilderFactory: StepBuilderFactory,
        bulkInsertReader: JpaCursorItemReader<Payment>
    ): Step =
        stepBuilderFactory["bulkInsertStep"]
            .chunk<Payment, Payment>(GLOBAL_CHUNK_SIZE)
            .reader(bulkInsertReader) // Reader는 JpaCursorItemReader 기반으로 조회
//            .writer(writerWithStatement) // Statement addBatch() 기반 저장 
            .writer(writerWithExposed) // Exposed 기반 저장
//            .writer(writerWithJpa) // JPA 기반 저장
            .build()

    @Bean
    @StepScope
    fun bulkInsertReader(
        entityManagerFactory: EntityManagerFactory
    ) = JpaCursorItemReaderBuilder<Payment>()
        .name("bulkInsertReader")
        .entityManagerFactory(entityManagerFactory)
        .queryString("SELECT p FROM Payment p")
        .build()


    private val writerWithStatement: ItemWriter<Payment> = ItemWriter { payments ->
        val sql = "insert into payment_back (amount, order_id) values (?, ?)"
        val connection = dataSource.connection
        val statement = connection.prepareStatement(sql)!!
        try {
            for (payment in payments) {
                statement.apply {
                    this.setBigDecimal(1, payment.amount)
                    this.setLong(2, payment.orderId)
                    this.addBatch()
                }
            }
            statement.executeBatch()
        } catch (e: Exception) {
            throw e
        } finally {
            if (statement.isClosed.not()) {
                statement.close()
            }
            if (connection.isClosed.not()) {
                connection.close()
            }
        }
    }

    private val writerWithJpa: ItemWriter<Payment> =
        ItemWriter { payments ->
            payments.map { payment ->
                PaymentBackJpa(
                    amount = payment.amount,
                    orderId = payment.orderId
                )
            }
                .also {
                    paymentBackJpaRepository.saveAll(it)
                }
        }


    private val writerWithExposed: ItemWriter<Payment> = ItemWriter { payments ->
        transaction(
            exposedDataBase
        ) {
            PaymentBack.batchInsert(
                data = payments,
                shouldReturnGeneratedValues = false
            ) { payment ->
                this[PaymentBack.orderId] = payment.orderId
                this[PaymentBack.amount] = payment.amount
            }
        }
    }
}

object PaymentBack : LongIdTable(name = "payment_back") {
    val amount = decimal("amount", 19, 4)
    val orderId = long("order_id")
}

transaction(
    exposedDataBase
) {
    PaymentBack.batchInsert(
        data = payments,
        shouldReturnGeneratedValues = false
    ) { payment ->
        this[PaymentBack.orderId] = payment.orderId
        this[PaymentBack.amount] = payment.amount
    }
}