SpringBatch QuerydslItemReader 라이브러리 만들기

...

	@Override
	@SuppressWarnings("unchecked")
	protected void doReadPage() {

		EntityTransaction tx = null;

		if (transacted) {
			tx = entityManager.getTransaction();
			tx.begin();

			entityManager.flush();
			entityManager.clear();
		} // end if

		Query query = createQuery().setFirstResult(getPage() * getPageSize()).setMaxResults(getPageSize());

		if (parameterValues != null) {
			for (Map.Entry<String, Object> me : parameterValues.entrySet()) {
				query.setParameter(me.getKey(), me.getValue());
			}
		}

		if (results == null) {
			results = new CopyOnWriteArrayList<>();
		}
		else {
			results.clear();
		}

		if (!transacted) {
			List<T> queryResult = query.getResultList();
			for (T entity : queryResult) {
				entityManager.detach(entity);
				results.add(entity);
			} // end if
		}
		else {
			results.addAll(query.getResultList());
			tx.commit();
		} // end if
	}

...

public class QueryDslPagingItemReader<T> extends AbstractPagingItemReader<T> {
    protected final Map<String, Object> jpaPropertyMap = new HashMap<>();
    protected EntityManagerFactory entityManagerFactory;
    protected EntityManager entityManager;
    protected Function<JPAQueryFactory, JPAQuery<T>> queryFunction;

    private boolean transacted = true;// default value

    public QueryDslPagingItemReader() {
        setName(ClassUtils.getShortName(QueryDslPagingItemReader.class));
    }

    public QueryDslPagingItemReader(EntityManagerFactory entityManagerFactory, int pageSize,
                                    Function<JPAQueryFactory, JPAQuery<T>> queryFunction) {
        this();
        this.entityManagerFactory = entityManagerFactory;
        this.queryFunction = queryFunction;
        setPageSize(pageSize);
    }

    /**
     * Create a query using an appropriate query provider (entityManager OR
     * queryProvider).
     */
    /**
     * 이 부분이 private이기 때문에 새롭게 작성하게 된 것! override할 수 없어서
     */
//    private Query createQuery() {
//        if (queryProvider == null) {
//            return entityManager.createQuery(queryString);
//        }
//        else {
//            return queryProvider.createQuery();
//        }
//    }
    /**
     * By default (true) the EntityTransaction will be started and committed around the
     * read. Can be overridden (false) in cases where the JPA implementation doesn't
     * support a particular transaction. (e.g. Hibernate with a JTA transaction). NOTE:
     * may cause problems in guaranteeing the object consistency in the
     * EntityManagerFactory.
     * @param transacted indicator
     */
    public void setTransacted(boolean transacted) {
        this.transacted = transacted;
    }

    @Override
    protected void doOpen() throws Exception {
        super.doOpen();

        entityManager = entityManagerFactory.createEntityManager(jpaPropertyMap);
        if (entityManager == null) {
            throw new DataAccessResourceFailureException("Unable to obtain an EntityManager");
        }
    }

    @Override
    @SuppressWarnings("unchecked")
    protected void doReadPage() {
        clearIfTransacted();

        JPQLQuery<T> query = createQuery()
                .offset(getPage() * getPageSize())
                .limit(getPageSize());

        initResults();

        fetchQuery(query);
    }

    protected void clearIfTransacted() {
        if (transacted) {
            entityManager.clear();
        }
    }

    protected JPAQuery<T> createQuery() {
        JPAQueryFactory queryFactory = new JPAQueryFactory(entityManager);
        return queryFunction.apply(queryFactory);
    }

    protected void initResults() {
        if (CollectionUtils.isEmpty(results)) {
            results = new ArrayList<>();
        } else {
            results.clear();
        }
    }

    protected void fetchQuery(JPQLQuery<T> query) {
        if (!transacted) {
            List<T> queryResult = query.fetch();
            for (T entity : queryResult) {
                entityManager.detach(entity);
                results.add(entity);
            }
        } else {
            results.addAll(query.fetch());
        }
    }

    @Override
    protected void doClose() throws Exception {
        entityManager.close();
        super.doClose();
    }
}

@Bean
public QueryDslPagingItemReader<CreateSalary> salaryReader() {
    return new QueryDslPagingItemReader<Salary>(entityManagerFactory, CHUNK_SIZE, queryFactory ->
             queryFactory.selectFrom(salary)
                     .where(salary.salaryId.eq(1L))
		);
}

public class QueryDslZeroPagingItemReader<T> extends QueryDslPagingItemReader<T> {

    public QueryDslZeroPagingItemReader() {
        super();
        setName(ClassUtils.getShortName(QueryDslZeroPagingItemReader.class));
    }

    public QueryDslZeroPagingItemReader(EntityManagerFactory entityManagerFactory,
                                        int pageSize,
                                        Function<JPAQueryFactory, JPAQuery<T>> queryFunction) {
        this();
        setTransacted(true);
        super.entityManagerFactory = entityManagerFactory;
        super.queryFunction = queryFunction;
        setPageSize(pageSize);
    }

    @Override
    @SuppressWarnings("unchecked")
    protected void doReadPage() {

        JPQLQuery<T> query = createQuery()
                .offset(0)
                .limit(getPageSize());

        initResults();

        fetchQuery(query);
    }
}

select *
FROM members
WHERE 조건문
ORDER BY id DESC
OFFSET 페이지 번호
LIMIT 페이지 사이즈 

SELECT *
FROM members as m
JOIN (
	SELECT id
	FROM members
	WHERE 조건문
	ORDER BY id DESC
	OFFSET 페이지번호
	LIMIT 페이지사이즈
) as temp 
ON temp.id = m.id

SELECT * 
FROM members
WHERE 조건문
AND id < 마지막조회 ID #직전 조회 결과의 마지막 id
ORDER BY id DESC
LIMIT 페이지 사이즈 

AND id < 마지막조회ID # 직전 조회 결과의 마지막 id
ORDER BY id DESC
LIMIT 페이지사이즈

public class QueryDslNoOffsetPagingItemReader<T> extends QueryDslPagingItemReader<T> {

    private QueryDslNoOffsetOptions<T> options;

    private QueryDslNoOffsetPagingItemReader() {
        super();
        setName(ClassUtils.getShortName(QueryDslNoOffsetPagingItemReader.class));
    }

    public QueryDslNoOffsetPagingItemReader(EntityManagerFactory entityManagerFactory,
                                            int pageSize,
                                            QueryDslNoOffsetOptions<T> options,
                                            Function<JPAQueryFactory, JPAQuery<T>> queryFunction) {
        super(entityManagerFactory, pageSize, queryFunction);
        setName(ClassUtils.getShortName(QueryDslNoOffsetPagingItemReader.class));
        this.options = options;
    }

    @Override
    @SuppressWarnings("unchecked")
    protected void doReadPage() {
        JPQLQuery<T> query = createQuery().limit(getPageSize());

        initResults();

        fetchQuery(query);

        resetCurrentIdIfNotLastPage();
    }

    @Override
    protected JPAQuery<T> createQuery() {
        JPAQueryFactory queryFactory = new JPAQueryFactory(entityManager);
        JPAQuery<T> query = queryFunction.apply(queryFactory);
        options.initKeys(query, getPage()); // 제일 첫번째 페이징시 시작해야할 ID 찾기

        return options.createQuery(query, getPage());
    }

    private void resetCurrentIdIfNotLastPage() {
        if (isNotEmptyResults()) {
            options.resetCurrentId(getLastItem());
        }
    }

    // 조회결과가 Empty이면 results에 null이 담긴다
    private boolean isNotEmptyResults() {
        return !CollectionUtils.isEmpty(results) && results.get(0) != null;
    }

    private T getLastItem() {
        return results.get(results.size() - 1);
    }
}

@Override
public void initKeys(JPAQuery<T> query, int page) {
    if(page == 0) {
        initFirstId(query);
        initLastId(query);

        log.debug("First Key= {}, Last Key = {}", currentId, lastId);
    }
}

@Override
protected void initFirstId(JPAQuery<T> query) {
    JPAQuery<T> clone = query.clone();
    boolean isGroupByQuery = isGroupByQuery(clone);

    if(isGroupByQuery) {
        currentId = clone
                .select(field)
                .orderBy(expression.isAsc()? field.asc() : field.desc())
                .fetchFirst();
    } else {
        currentId = clone
                .select(expression.isAsc()? field.min(): field.max())
                .fetchFirst();
    }

}

@Override
protected void initLastId(JPAQuery<T> query) {
    JPAQuery<T> clone = query.clone();
    boolean isGroupByQuery = isGroupByQuery(clone);

    if(isGroupByQuery) {
        lastId = clone
                .select(field)
                .orderBy(expression.isAsc()? field.desc() : field.asc())
                .fetchFirst();
    } else {
        lastId = clone
                .select(expression.isAsc()? field.max(): field.min())
                .fetchFirst();
    }
}

@Override
public JPAQuery<T> createQuery(JPAQuery<T> query, int page) {
    if(currentId == null) {
        return query;
    }

    return query
            .where(whereExpression(page))
            .orderBy(orderExpression());
}

private BooleanExpression whereExpression(int page) {
    return expression.where(field, page, currentId)
            .and(expression.isAsc()? field.loe(lastId) : field.goe(lastId));
}

private OrderSpecifier<N> orderExpression() {
    return expression.order(field);
}

//---------------------- 아래는 추상클래스 구현 부분 ------------------------

public boolean isGroupByQuery(JPAQuery<T> query) {
    return isGroupByQuery(query.toString());
}

public boolean isGroupByQuery(String sql) {
    return sql.contains("group by");
}