vttranlina commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1855051965
##########
mailbox/postgres/src/main/java/org/apache/james/mailbox/postgres/mail/dao/PostgresMailboxMessageDAO.java:
##########
@@ -117,78 +115,110 @@ public PostgresMailboxMessageDAO create(Optional<Domain>
domain) {
public static final SortField<Long> DEFAULT_SORT_ORDER_BY =
MESSAGE_UID.asc();
- private static SelectFinalStep<Record1<Long>>
selectMessageUidByMailboxIdAndExtraConditionQuery(PostgresMailboxId mailboxId,
Condition extraCondition, Limit limit, DSLContext dslContext) {
- SelectSeekStep1<Record1<Long>, Long> queryWithoutLimit =
dslContext.select(MESSAGE_UID)
- .from(TABLE_NAME)
- .where(MAILBOX_ID.eq((mailboxId.asUuid())))
- .and(extraCondition)
- .orderBy(MESSAGE_UID.asc());
- return limit.getLimit().map(limitValue ->
(SelectFinalStep<Record1<Long>>) queryWithoutLimit.limit(limitValue))
- .orElse(queryWithoutLimit);
- }
-
private final PostgresExecutor postgresExecutor;
+ private final int queryBatchSize;
- public PostgresMailboxMessageDAO(PostgresExecutor postgresExecutor) {
+ public PostgresMailboxMessageDAO(PostgresExecutor postgresExecutor,
+ int queryBatchSize) {
this.postgresExecutor = postgresExecutor;
+ this.queryBatchSize = queryBatchSize;
}
public Mono<MessageUid> findFirstUnseenMessageUid(PostgresMailboxId
mailboxId) {
- return postgresExecutor.executeRow(dslContext ->
Mono.from(selectMessageUidByMailboxIdAndExtraConditionQuery(mailboxId,
- IS_SEEN.eq(false), Limit.limit(1), dslContext)))
+ return postgresExecutor.executeRow(dslContext ->
Flux.from(dslContext.select(MESSAGE_UID)
+ .from(TABLE_NAME)
+ .where(MAILBOX_ID.eq((mailboxId.asUuid())))
+ .and(IS_SEEN.eq(false))
+ .orderBy(DEFAULT_SORT_ORDER_BY)
+ .limit(1)))
.map(RECORD_TO_MESSAGE_UID_FUNCTION);
}
public Flux<MessageUid> listUnseen(PostgresMailboxId mailboxId) {
- return postgresExecutor.executeRows(dslContext ->
Flux.from(selectMessageUidByMailboxIdAndExtraConditionQuery(mailboxId,
- IS_SEEN.eq(false), Limit.unlimited(), dslContext)))
- .map(RECORD_TO_MESSAGE_UID_FUNCTION);
+ return Flux.range(0, OFFSET_UNLIMITED)
+ .concatMap(offsetIndex -> listUnseenBatch(mailboxId, offsetIndex *
queryBatchSize, queryBatchSize))
+ .takeUntil(List::isEmpty)
+ .flatMapIterable(Function.identity());
+ }
+
+ private Mono<List<MessageUid>> listUnseenBatch(PostgresMailboxId
mailboxId, int offset, int batchSize) {
+ return postgresExecutor.executeRows(dslContext ->
Flux.from(dslContext.select(MESSAGE_UID)
Review Comment:
That is also my concern.
`memory resource` (only uid) is not only that we must care. when the number
of row is large, the batch help we early release connection to pool, for next
round.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@james.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscr...@james.apache.org
For additional commands, e-mail: notifications-h...@james.apache.org
