You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by "vttranlina (via GitHub)" <gi...@apache.org> on 2024/11/20 09:19:47 UTC
[PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 [james-project]
vttranlina opened a new pull request, #2514:
URL: https://github.com/apache/james-project/pull/2514
(no comment)
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1858068476
##########
backends-common/postgres/src/main/java/org/apache/james/backends/postgres/PostgresConfiguration.java:
##########
@@ -59,6 +59,8 @@ public class PostgresConfiguration {
public static final String SSL_MODE_DEFAULT_VALUE = "allow";
public static final String JOOQ_REACTIVE_TIMEOUT = "jooq.reactive.timeout";
public static final Duration JOOQ_REACTIVE_TIMEOUT_DEFAULT_VALUE = Duration.ofSeconds(10);
+ public static final String QUERY_BATCH_SIZE = "query.batch.size";
+ public static final int QUERY_BATCH_SIZE_DEFAULT_VALUE = 5000;
Review Comment:
Use a system property and default to 5000
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1853560885
##########
server/apps/postgres-app/sample-configuration/postgres.properties:
##########
@@ -42,4 +42,7 @@ by-pass-rls.pool.max.size=10
ssl.mode=allow
## Duration. Optional, defaults to 10 second. jOOQ reactive timeout when executing Postgres query. This setting prevent jooq reactive bug from causing hanging issue.
-#jooq.reactive.timeout=10second
\ No newline at end of file
+#jooq.reactive.timeout=10second
+
+# Integer. Optional, defaults to 100. In case of large data, this argument specifies the maximum number of rows to return in a single batch set when executing query.
+#jooq.reactive.timeout=100
Review Comment:
Oups copy n paster mistake?
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2503994100
Please squash
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "vttranlina (via GitHub)" <gi...@apache.org>.
vttranlina commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2496575416
Fixup `.collectList` first for CI checking
I'm processing
> Let's look for bugs: can we mark 1000 messages as deleted and then expunge and ensure we did delete then all?
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "vttranlina (via GitHub)" <gi...@apache.org>.
vttranlina commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1859850515
##########
mailbox/postgres/src/main/java/org/apache/james/mailbox/postgres/mail/dao/PostgresMailboxMessageDAO.java:
##########
@@ -117,78 +116,107 @@ 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;
public PostgresMailboxMessageDAO(PostgresExecutor postgresExecutor) {
this.postgresExecutor = postgresExecutor;
}
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 * BATCH_SIZE, BATCH_SIZE))
+ .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)
+ .from(TABLE_NAME)
+ .where(MAILBOX_ID.eq((mailboxId.asUuid())))
+ .and(IS_SEEN.eq(false))
+ .orderBy(DEFAULT_SORT_ORDER_BY)
+ .limit(batchSize)
+ .offset(offset)))
+ .map(RECORD_TO_MESSAGE_UID_FUNCTION)
+ .collectList()
Review Comment:
https://github.com/apache/james-project/pull/2514#issuecomment-2490348291
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 [james-project]
Posted by "vttranlina (via GitHub)" <gi...@apache.org>.
vttranlina commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2490210967
Although using a connection pool is a "application layer" solution that can fix this issue, but IMO, it does not solve the root cause.
The connection pool size must be "large enough" to avoid errors.
This can be better understood through the following example:
Give: Connection pool size = 3
When reactive chain logic:
```
getUser
.flatMap(user -> getMailbox)
.flatMap(mailbox -> getMessages)
.flatMap(message -> updateQuota)
```
You can see in the chain that we need 4 connections available at the same time.
4 > 3 => We will get an exception.
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2489043767
WOOT!
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2490908310
> I have an idea: If a connection can be used safely by different entities in the same thread, we can share connections per thread -> reduce number of connections.
Won't play well in a reactive context
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "Arsnael (via GitHub)" <gi...@apache.org>.
Arsnael merged PR #2514:
URL: https://github.com/apache/james-project/pull/2514
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "vttranlina (via GitHub)" <gi...@apache.org>.
vttranlina commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1858093810
##########
server/blob/blob-postgres/src/test/java/org/apache/james/blob/postgres/PostgresBlobStoreGCAlgorithmTest.java:
##########
@@ -0,0 +1,44 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ ****************************************************************/
+
+package org.apache.james.blob.postgres;
+
+import org.apache.james.backends.postgres.PostgresExtension;
+import org.apache.james.blob.api.BlobStoreDAO;
+import org.apache.james.blob.api.PlainBlobId;
+import org.apache.james.server.blob.deduplication.BloomFilterGCAlgorithmContract;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+public class PostgresBlobStoreGCAlgorithmTest implements BloomFilterGCAlgorithmContract {
+
+ @RegisterExtension
+ static PostgresExtension postgresExtension = PostgresExtension.withoutRowLevelSecurity(PostgresBlobStorageModule.MODULE, PostgresExtension.PoolSize.LARGE);
+ private PostgresBlobStoreDAO blobStore;
+
+ @BeforeEach
+ void beforeEach() {
+ blobStore = new PostgresBlobStoreDAO(postgresExtension.getDefaultPostgresExecutor(), new PlainBlobId.Factory(), 1000);
+ }
+
+ @Override
+ public BlobStoreDAO blobStoreDAO() {
+ return blobStore;
+ }
Review Comment:
it failed, I tested it
The reason as you write at https://github.com/apache/james-project/pull/2514#discussion_r1853900136
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 [james-project]
Posted by "vttranlina (via GitHub)" <gi...@apache.org>.
vttranlina commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1849926161
##########
backends-common/postgres/src/main/java/org/apache/james/backends/postgres/utils/PostgresExecutor.java:
##########
@@ -145,8 +143,6 @@ public Flux<Record> executeDeleteAndReturnList(Function<DSLContext, DeleteResult
.flatMapMany(queryFunction)
.timeout(postgresConfiguration.getJooqReactiveTimeout())
.doOnError(TimeoutException.class, e -> LOGGER.error(JOOQ_TIMEOUT_ERROR_LOG, e))
- .collectList()
Review Comment:
The hanging issue occurs when connection-A is processing a query, and another where of the code tries to use connection-A for a different query.
I guess this issue might be resolved when we using a connection pool (implemented)
Wait to CI result for confirmation
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2490290568
(Indeed)
ChatGPT chat if relevant: this *might* be an actuall PG limitation we hit...
```
Is a SQL connection usable to run other queries if I am paging results from the database on a select query?
```
Whether a SQL connection can be used to run other queries while paging results from a database depends on the specifics of how the database and the connection are configured, as well as the SQL client library you're using. Here are the key considerations:
### 1. **Cursor Behavior**
- **Forward-Only Cursors**: If you're using a forward-only cursor for paging (common in libraries such as JDBC or Python's `psycopg2`), the connection is typically locked for that query, and running another query might not be allowed until the current one completes or the cursor is closed.
- **Scrollable Cursors**: Some databases support scrollable cursors, which might allow other queries on the same connection depending on the database's implementation.
### 2. **Database Locking and Transaction Scope**
- If the paging query is running within a transaction, the database may hold locks that prevent concurrent queries on the same connection.
- Running multiple queries simultaneously on a single connection could also violate transactional guarantees.
### 3. **Database Support for Concurrent Queries**
- Some databases (e.g., PostgreSQL, MySQL) support only one active query per connection at a time. Running another query will result in an error unless the first query is completed or the results are fully fetched.
- Databases like SQL Server and Oracle allow multiple active result sets (MARS) or similar mechanisms, enabling simultaneous queries under certain conditions.
### 4. **Library-Specific Considerations**
- Some client libraries (like Python's `asyncpg` for PostgreSQL) allow multiple queries to be queued asynchronously on a single connection.
- Libraries like `sqlalchemy` in Python often manage connection pooling, enabling multiple queries across multiple connections, even if one connection is busy.
### Best Practices for Paging
- **Use Connection Pooling**: Tools like HikariCP (Java) or SQLAlchemy’s connection pool ensure each query can use its own connection, avoiding contention.
- **Fetch Small Chunks**: Retrieve only the required number of rows for each page to minimize connection usage.
- **Close Cursors**: Ensure cursors are closed promptly after fetching the results.
### Conclusion
If you're paging results from a database on a select query, it's generally safer to assume that the connection is not available for other queries until the operation completes. Using connection pooling or multiple connections can mitigate this limitation effectively.
```
Does postgresql support scrollable cursors?
Are scrollable cursors implemented with R2DC library?
```
### PostgreSQL and Scrollable Cursors
**1. PostgreSQL Native Support:**
PostgreSQL does support scrollable cursors. A scrollable cursor allows you to fetch rows in any order (e.g., forward, backward, relative, absolute positioning). This is particularly useful for applications that need random access to query results.
- To create a scrollable cursor in PostgreSQL, you use the SQL `DECLARE` statement with the `SCROLL` keyword:
```sql
DECLARE my_cursor SCROLL CURSOR FOR SELECT * FROM my_table;
```
- Once the cursor is declared, you can use `FETCH` commands like `FETCH NEXT`, `FETCH PRIOR`, or `FETCH ABSOLUTE` to navigate through the result set.
- Scrollable cursors are limited to explicit transactions. That means you need to wrap them within a `BEGIN`/`COMMIT` or `BEGIN`/`ROLLBACK` block.
---
**2. R2DBC and Scrollable Cursors:**
[R2DBC](https://r2dbc.io/) (Reactive Relational Database Connectivity) is designed for reactive programming and focuses on streaming data asynchronously.
- R2DBC itself does not directly expose scrollable cursors in the way traditional JDBC does. Instead, it provides reactive streams for processing results row-by-row in a non-blocking manner.
- While PostgreSQL supports scrollable cursors natively, the R2DBC PostgreSQL driver does not provide a direct interface for them. You would need to implement a workaround using PostgreSQL's native SQL commands for cursor operations.
---
### Summary
- **PostgreSQL:** Fully supports scrollable cursors with SQL commands.
- **R2DBC:** Does not natively implement scrollable cursors. However, its reactive streams can achieve similar functionality by processing and re-querying data as needed for reactive workflows.
If scrollable cursors are a strict requirement for your application, traditional JDBC might be a better fit than R2DBC.
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "vttranlina (via GitHub)" <gi...@apache.org>.
vttranlina commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2505198989
# Performance test result
When running performance tests for the latest code, I observed a degradation in some result index compared to the previous run. Suspecting potential system infrastructure instability, I decided to rerun the performance tests for both the "before" and "after" states of the PR.
The detailed results are shown below.
## Gatling
## Metrics
- Before
```
# TYPE postgres_execution summary
postgres_execution{quantile="0.5",} 0.0019005430000000002
postgres_execution{quantile="0.75",} 0.0029982710000000003
postgres_execution{quantile="0.95",} 0.008323071000000001
postgres_execution{quantile="0.98",} 0.014811135000000001
postgres_execution{quantile="0.99",} 0.024379391
postgres_execution{quantile="0.999",} 0.071827455
postgres_execution_count 2149596.0
```
- After
```
# TYPE postgres_execution summary
postgres_execution{quantile="0.5",} 0.0015646710000000001
postgres_execution{quantile="0.75",} 0.002260991
postgres_execution{quantile="0.95",} 0.005013503
postgres_execution{quantile="0.98",} 0.007438335000000001
postgres_execution{quantile="0.99",} 0.009961471000000001
postgres_execution{quantile="0.999",} 0.028311551
postgres_execution_count 2244489.0
```
## pg_stat_statement
- Before
- After
File test result: https://github.com/vttranlina/vttranlina.github.io/tree/main/gatling_20241127
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1853892580
##########
mailbox/postgres/src/test/java/org/apache/james/mailbox/postgres/search/AllSearchOverrideTest.java:
##########
@@ -51,7 +51,7 @@ public class AllSearchOverrideTest {
@BeforeEach
void setUp() {
postgresMessageDAO = new PostgresMessageDAO(postgresExtension.getDefaultPostgresExecutor(), new PlainBlobId.Factory());
- postgresMailboxMessageDAO = new PostgresMailboxMessageDAO(postgresExtension.getDefaultPostgresExecutor());
+ postgresMailboxMessageDAO = new PostgresMailboxMessageDAO(postgresExtension.getDefaultPostgresExecutor(), postgresExtension.getExecutorFactory().queryBatchSize());
Review Comment:
for all search overrides we only load a list of uids. I think .collectList would be better. At least we could start with it.
##########
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:
For uid only let's .collectList and not do application side batching.
##########
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)
+ .from(TABLE_NAME)
+ .where(MAILBOX_ID.eq((mailboxId.asUuid())))
+ .and(IS_SEEN.eq(false))
+ .orderBy(DEFAULT_SORT_ORDER_BY)
+ .limit(batchSize)
+ .offset(offset)))
+ .map(RECORD_TO_MESSAGE_UID_FUNCTION)
+ .collectList()
+ .switchIfEmpty(Mono.just(ImmutableList.of()));
}
public Flux<MessageUid> listUnseen(PostgresMailboxId mailboxId, MessageRange range) {
- switch (range.getType()) {
- case ALL:
- return listUnseen(mailboxId);
- case FROM:
- return postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.select(MESSAGE_UID)
- .from(TABLE_NAME)
- .where(MAILBOX_ID.eq(mailboxId.asUuid()))
- .and(IS_SEEN.eq(false))
- .and(MESSAGE_UID.greaterOrEqual(range.getUidFrom().asLong()))
- .orderBy(DEFAULT_SORT_ORDER_BY)))
- .map(RECORD_TO_MESSAGE_UID_FUNCTION);
- case RANGE:
- return postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.select(MESSAGE_UID)
- .from(TABLE_NAME)
- .where(MAILBOX_ID.eq(mailboxId.asUuid()))
- .and(IS_SEEN.eq(false))
- .and(MESSAGE_UID.greaterOrEqual(range.getUidFrom().asLong()))
- .and(MESSAGE_UID.lessOrEqual(range.getUidTo().asLong()))
- .orderBy(DEFAULT_SORT_ORDER_BY)))
- .map(RECORD_TO_MESSAGE_UID_FUNCTION);
- case ONE:
- return postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.select(MESSAGE_UID)
- .from(TABLE_NAME)
- .where(MAILBOX_ID.eq(mailboxId.asUuid()))
- .and(IS_SEEN.eq(false))
- .and(MESSAGE_UID.eq(range.getUidFrom().asLong()))
- .orderBy(DEFAULT_SORT_ORDER_BY)))
- .map(RECORD_TO_MESSAGE_UID_FUNCTION);
- default:
- throw new RuntimeException("Unsupported range type " + range.getType());
- }
+ return switch (range.getType()) {
+ case ALL -> listUnseen(mailboxId);
+ case FROM -> postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.select(MESSAGE_UID)
+ .from(TABLE_NAME)
+ .where(MAILBOX_ID.eq(mailboxId.asUuid()))
+ .and(IS_SEEN.eq(false))
+ .and(MESSAGE_UID.greaterOrEqual(range.getUidFrom().asLong()))
+ .orderBy(DEFAULT_SORT_ORDER_BY)), EAGER_FETCH)
+ .map(RECORD_TO_MESSAGE_UID_FUNCTION);
+ case RANGE -> postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.select(MESSAGE_UID)
+ .from(TABLE_NAME)
+ .where(MAILBOX_ID.eq(mailboxId.asUuid()))
+ .and(IS_SEEN.eq(false))
+ .and(MESSAGE_UID.greaterOrEqual(range.getUidFrom().asLong()))
+ .and(MESSAGE_UID.lessOrEqual(range.getUidTo().asLong()))
+ .orderBy(DEFAULT_SORT_ORDER_BY)), EAGER_FETCH)
+ .map(RECORD_TO_MESSAGE_UID_FUNCTION);
+ case ONE -> postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.select(MESSAGE_UID)
+ .from(TABLE_NAME)
+ .where(MAILBOX_ID.eq(mailboxId.asUuid()))
+ .and(IS_SEEN.eq(false))
+ .and(MESSAGE_UID.eq(range.getUidFrom().asLong()))
+ .orderBy(DEFAULT_SORT_ORDER_BY)), EAGER_FETCH)
+ .map(RECORD_TO_MESSAGE_UID_FUNCTION);
+ };
}
public Flux<MessageUid> findAllRecentMessageUid(PostgresMailboxId mailboxId) {
Review Comment:
Same
##########
mailbox/postgres/src/main/java/org/apache/james/mailbox/postgres/mail/dao/PostgresMailboxMessageDAO.java:
##########
@@ -345,33 +425,66 @@ public Flux<SimpleMailboxMessage.Builder> findMessagesByMailboxIdAndUIDs(Postgre
}
public Flux<MessageUid> findDeletedMessagesByMailboxId(PostgresMailboxId mailboxId) {
+ return Flux.range(0, OFFSET_UNLIMITED)
Review Comment:
Idem we are selecting uid only so let's rather .collectList ?
##########
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)
+ .from(TABLE_NAME)
+ .where(MAILBOX_ID.eq((mailboxId.asUuid())))
+ .and(IS_SEEN.eq(false))
+ .orderBy(DEFAULT_SORT_ORDER_BY)
+ .limit(batchSize)
+ .offset(offset)))
+ .map(RECORD_TO_MESSAGE_UID_FUNCTION)
+ .collectList()
+ .switchIfEmpty(Mono.just(ImmutableList.of()));
}
public Flux<MessageUid> listUnseen(PostgresMailboxId mailboxId, MessageRange range) {
- switch (range.getType()) {
- case ALL:
- return listUnseen(mailboxId);
- case FROM:
- return postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.select(MESSAGE_UID)
- .from(TABLE_NAME)
- .where(MAILBOX_ID.eq(mailboxId.asUuid()))
- .and(IS_SEEN.eq(false))
- .and(MESSAGE_UID.greaterOrEqual(range.getUidFrom().asLong()))
- .orderBy(DEFAULT_SORT_ORDER_BY)))
- .map(RECORD_TO_MESSAGE_UID_FUNCTION);
- case RANGE:
- return postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.select(MESSAGE_UID)
- .from(TABLE_NAME)
- .where(MAILBOX_ID.eq(mailboxId.asUuid()))
- .and(IS_SEEN.eq(false))
- .and(MESSAGE_UID.greaterOrEqual(range.getUidFrom().asLong()))
- .and(MESSAGE_UID.lessOrEqual(range.getUidTo().asLong()))
- .orderBy(DEFAULT_SORT_ORDER_BY)))
- .map(RECORD_TO_MESSAGE_UID_FUNCTION);
- case ONE:
- return postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.select(MESSAGE_UID)
- .from(TABLE_NAME)
- .where(MAILBOX_ID.eq(mailboxId.asUuid()))
- .and(IS_SEEN.eq(false))
- .and(MESSAGE_UID.eq(range.getUidFrom().asLong()))
- .orderBy(DEFAULT_SORT_ORDER_BY)))
- .map(RECORD_TO_MESSAGE_UID_FUNCTION);
- default:
- throw new RuntimeException("Unsupported range type " + range.getType());
- }
+ return switch (range.getType()) {
+ case ALL -> listUnseen(mailboxId);
+ case FROM -> postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.select(MESSAGE_UID)
+ .from(TABLE_NAME)
+ .where(MAILBOX_ID.eq(mailboxId.asUuid()))
+ .and(IS_SEEN.eq(false))
+ .and(MESSAGE_UID.greaterOrEqual(range.getUidFrom().asLong()))
+ .orderBy(DEFAULT_SORT_ORDER_BY)), EAGER_FETCH)
+ .map(RECORD_TO_MESSAGE_UID_FUNCTION);
+ case RANGE -> postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.select(MESSAGE_UID)
+ .from(TABLE_NAME)
+ .where(MAILBOX_ID.eq(mailboxId.asUuid()))
+ .and(IS_SEEN.eq(false))
+ .and(MESSAGE_UID.greaterOrEqual(range.getUidFrom().asLong()))
+ .and(MESSAGE_UID.lessOrEqual(range.getUidTo().asLong()))
+ .orderBy(DEFAULT_SORT_ORDER_BY)), EAGER_FETCH)
+ .map(RECORD_TO_MESSAGE_UID_FUNCTION);
+ case ONE -> postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.select(MESSAGE_UID)
+ .from(TABLE_NAME)
+ .where(MAILBOX_ID.eq(mailboxId.asUuid()))
+ .and(IS_SEEN.eq(false))
+ .and(MESSAGE_UID.eq(range.getUidFrom().asLong()))
+ .orderBy(DEFAULT_SORT_ORDER_BY)), EAGER_FETCH)
+ .map(RECORD_TO_MESSAGE_UID_FUNCTION);
+ };
}
public Flux<MessageUid> findAllRecentMessageUid(PostgresMailboxId mailboxId) {
- return postgresExecutor.executeRows(dslContext -> Flux.from(selectMessageUidByMailboxIdAndExtraConditionQuery(mailboxId,
- IS_RECENT.eq(true), Limit.unlimited(), dslContext)))
- .map(RECORD_TO_MESSAGE_UID_FUNCTION);
+ return Flux.range(0, OFFSET_UNLIMITED)
+ .concatMap(offsetIndex -> findAllRecentMessageUidBatch(mailboxId, offsetIndex * queryBatchSize, queryBatchSize))
+ .takeUntil(List::isEmpty)
+ .flatMapIterable(Function.identity());
+ }
+
+ private Mono<List<MessageUid>> findAllRecentMessageUidBatch(PostgresMailboxId mailboxId, int offset, int batchSize) {
+ return postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.select(MESSAGE_UID)
+ .from(TABLE_NAME)
+ .where(MAILBOX_ID.eq((mailboxId.asUuid())))
+ .and(IS_RECENT.eq(true))
+ .orderBy(DEFAULT_SORT_ORDER_BY)
+ .limit(batchSize)
+ .offset(offset)))
+ .map(RECORD_TO_MESSAGE_UID_FUNCTION)
+ .collectList()
+ .switchIfEmpty(Mono.just(ImmutableList.of()));
}
public Flux<MessageUid> listAllMessageUid(PostgresMailboxId mailboxId) {
- return postgresExecutor.executeRows(dslContext -> Flux.from(selectMessageUidByMailboxIdAndExtraConditionQuery(mailboxId,
- DSL.noCondition(), Limit.unlimited(), dslContext)))
- .map(RECORD_TO_MESSAGE_UID_FUNCTION);
+ return Flux.range(0, OFFSET_UNLIMITED)
+ .concatMap(offsetIndex -> listAllMessageUidBatch(mailboxId, offsetIndex * queryBatchSize, queryBatchSize))
+ .takeUntil(List::isEmpty)
+ .flatMapIterable(Function.identity());
+ }
+
+ private Mono<List<MessageUid>> listAllMessageUidBatch(PostgresMailboxId mailboxId, int offset, int batchSize) {
Review Comment:
Same
##########
backends-common/postgres/src/main/java/org/apache/james/backends/postgres/PostgresConfiguration.java:
##########
@@ -59,6 +59,8 @@ public class PostgresConfiguration {
public static final String SSL_MODE_DEFAULT_VALUE = "allow";
public static final String JOOQ_REACTIVE_TIMEOUT = "jooq.reactive.timeout";
public static final Duration JOOQ_REACTIVE_TIMEOUT_DEFAULT_VALUE = Duration.ofSeconds(10);
+ public static final String QUERY_BATCH_SIZE = "query.batch.size";
+ public static final int QUERY_BATCH_SIZE_DEFAULT_VALUE = 100;
Review Comment:
let's do atleast 5000 (cassandra batch size)
##########
mailbox/postgres/src/main/java/org/apache/james/mailbox/postgres/mail/dao/PostgresMailboxDAO.java:
##########
@@ -202,11 +204,18 @@ public Flux<PostgresMailbox> findMailboxByIds(List<PostgresMailboxId> mailboxIds
public Flux<PostgresMailbox> findMailboxWithPathLike(MailboxQuery.UserBound query) {
String pathLike = MailboxExpressionBackwardCompatibility.getPathLike(query);
+ Function<MailboxQuery.UserBound, Condition> getQueryCondition = mailboxQuery -> {
Review Comment:
I expect the count of mailboxes to be small (per user) few thousands at best.
Let's just .collectList ?
##########
server/blob/blob-postgres/src/main/java/org/apache/james/blob/postgres/PostgresBlobStoreDAO.java:
##########
@@ -151,9 +165,20 @@ public Flux<BucketName> listBuckets() {
@Override
public Flux<BlobId> listBlobs(BucketName bucketName) {
+ return Flux.range(0, OFFSET_UNLIMITED)
+ .concatMap(offsetIndex -> listBlobsBatch(bucketName, offsetIndex * queryBatchSize, queryBatchSize))
+ .takeUntil(List::isEmpty)
+ .flatMapIterable(Function.identity());
+ }
+
+ private Mono<List<BlobId>> listBlobsBatch(BucketName bucketName, int offset, int batchSize) {
return postgresExecutor.executeRows(dsl -> Flux.from(dsl.select(BLOB_ID)
.from(TABLE_NAME)
- .where(BUCKET_NAME.eq(bucketName.asString()))))
- .map(record -> blobIdFactory.parse(record.get(BLOB_ID)));
+ .where(BUCKET_NAME.eq(bucketName.asString()))
+ .limit(batchSize)
+ .offset(offset)))
Review Comment:
We are missing ordering to ensure the stability of the transformation.
Also here we have a massive issue: the use case is to delete as we iterate. This means that, without snapshot reads, the writes would impact the reads and cause entries to be skept.
(I predict GC with more blobs than batch size NOT to work well.
We are a bit stuck here. I'd say only small size installs would b okay storing blobs in PG so we *could* get over it by just doing a collectList - serious deployments with millions of blobs will anyway choose S3.
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "vttranlina (via GitHub)" <gi...@apache.org>.
vttranlina commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1855796686
##########
server/blob/blob-postgres/src/main/java/org/apache/james/blob/postgres/PostgresBlobStoreDAO.java:
##########
@@ -151,9 +165,20 @@ public Flux<BucketName> listBuckets() {
@Override
public Flux<BlobId> listBlobs(BucketName bucketName) {
+ return Flux.range(0, OFFSET_UNLIMITED)
+ .concatMap(offsetIndex -> listBlobsBatch(bucketName, offsetIndex * queryBatchSize, queryBatchSize))
+ .takeUntil(List::isEmpty)
+ .flatMapIterable(Function.identity());
+ }
+
+ private Mono<List<BlobId>> listBlobsBatch(BucketName bucketName, int offset, int batchSize) {
return postgresExecutor.executeRows(dsl -> Flux.from(dsl.select(BLOB_ID)
.from(TABLE_NAME)
- .where(BUCKET_NAME.eq(bucketName.asString()))))
- .map(record -> blobIdFactory.parse(record.get(BLOB_ID)));
+ .where(BUCKET_NAME.eq(bucketName.asString()))
+ .limit(batchSize)
+ .offset(offset)))
Review Comment:
> We are missing ordering to ensure the stability of the transformation.
updated
> Also here we have a massive issue: the use case is to delete as we iterate. This means that, without snapshot reads, the writes would impact the reads and cause entries to be skept.
Indeed, that's correct. However, this is a GC feature. While batch commands may not result in thorough cleanup initially, with repeated GC runs, the data will gradually become more accurate.
What do you think? `collectList` or NOT?
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1858103002
##########
server/blob/blob-postgres/src/test/java/org/apache/james/blob/postgres/PostgresBlobStoreGCAlgorithmTest.java:
##########
@@ -0,0 +1,44 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ ****************************************************************/
+
+package org.apache.james.blob.postgres;
+
+import org.apache.james.backends.postgres.PostgresExtension;
+import org.apache.james.blob.api.BlobStoreDAO;
+import org.apache.james.blob.api.PlainBlobId;
+import org.apache.james.server.blob.deduplication.BloomFilterGCAlgorithmContract;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+public class PostgresBlobStoreGCAlgorithmTest implements BloomFilterGCAlgorithmContract {
+
+ @RegisterExtension
+ static PostgresExtension postgresExtension = PostgresExtension.withoutRowLevelSecurity(PostgresBlobStorageModule.MODULE, PostgresExtension.PoolSize.LARGE);
+ private PostgresBlobStoreDAO blobStore;
+
+ @BeforeEach
+ void beforeEach() {
+ blobStore = new PostgresBlobStoreDAO(postgresExtension.getDefaultPostgresExecutor(), new PlainBlobId.Factory(), 1000);
+ }
+
+ @Override
+ public BlobStoreDAO blobStoreDAO() {
+ return blobStore;
+ }
Review Comment:
Ok then eager fetch here please!
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 [james-project]
Posted by "vttranlina (via GitHub)" <gi...@apache.org>.
vttranlina commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2490748107
> If a connection can be used safely by different entities in the same thread
"entities", what your mean?
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 [james-project]
Posted by "hungphan227 (via GitHub)" <gi...@apache.org>.
hungphan227 commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2490679888
I have an idea: If a connection can be used safely by different entities in the same thread, we can share connections per thread -> reduce number of connections.
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1858103002
##########
server/blob/blob-postgres/src/test/java/org/apache/james/blob/postgres/PostgresBlobStoreGCAlgorithmTest.java:
##########
@@ -0,0 +1,44 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ ****************************************************************/
+
+package org.apache.james.blob.postgres;
+
+import org.apache.james.backends.postgres.PostgresExtension;
+import org.apache.james.blob.api.BlobStoreDAO;
+import org.apache.james.blob.api.PlainBlobId;
+import org.apache.james.server.blob.deduplication.BloomFilterGCAlgorithmContract;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+public class PostgresBlobStoreGCAlgorithmTest implements BloomFilterGCAlgorithmContract {
+
+ @RegisterExtension
+ static PostgresExtension postgresExtension = PostgresExtension.withoutRowLevelSecurity(PostgresBlobStorageModule.MODULE, PostgresExtension.PoolSize.LARGE);
+ private PostgresBlobStoreDAO blobStore;
+
+ @BeforeEach
+ void beforeEach() {
+ blobStore = new PostgresBlobStoreDAO(postgresExtension.getDefaultPostgresExecutor(), new PlainBlobId.Factory(), 1000);
+ }
+
+ @Override
+ public BlobStoreDAO blobStoreDAO() {
+ return blobStore;
+ }
Review Comment:
Ok then eager fetch here please!
(and commit the test)
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "hungphan227 (via GitHub)" <gi...@apache.org>.
hungphan227 commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1858597631
##########
mailbox/postgres/src/main/java/org/apache/james/mailbox/postgres/mail/dao/PostgresMailboxMessageDAO.java:
##########
@@ -117,78 +116,107 @@ 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;
public PostgresMailboxMessageDAO(PostgresExecutor postgresExecutor) {
this.postgresExecutor = postgresExecutor;
}
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 * BATCH_SIZE, BATCH_SIZE))
+ .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)
+ .from(TABLE_NAME)
+ .where(MAILBOX_ID.eq((mailboxId.asUuid())))
+ .and(IS_SEEN.eq(false))
+ .orderBy(DEFAULT_SORT_ORDER_BY)
+ .limit(batchSize)
+ .offset(offset)))
+ .map(RECORD_TO_MESSAGE_UID_FUNCTION)
+ .collectList()
Review Comment:
We remove collectList but still use it in some cases. Really complicated to me.
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "vttranlina (via GitHub)" <gi...@apache.org>.
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "vttranlina (via GitHub)" <gi...@apache.org>.
vttranlina commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2495274403
> Let's look for bugs: can we mark 1000 messages as deleted and then expunge and ensure we did delete then all?
> Also keep in mind 'application batching' needs sort order in order to be deterministic.
👍
> Is there other use cases where lack of snapshot reads really hurt?
> Can we think of tasks iterating messages and deleting them?
Do you mean that during a batch query, the data might change, causing the query result to become outdated?
And potentially, the result of the next batch pagination query could be incorrect (especially in the case of deletions, as it changes the total size)?
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2490348291
@vttranlina does JOOQ offers some way to do lazy fetching in a reactive fashion and does it help?
Discussed with @yadd:
```
execute(select().from(MESSAGES))
```
Is syntactically equivalent to:
```
// Execute as many time as needed
execute(select().from(MESSAGES))
.sortOn(UID)
.where(UID > highestUID so far)
.limit(pageSize)
```
IE given a mailbox with 2500 messages we execute the select 3 times.
WHat it effectively does in implement paging within the application.
We could:
- .collectList when working with users, user mailboxes and maybe even mailboxes. COunt of entity is low.
- However maybe messages would be the only place where such kind of fetching really make sense ?
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1860104884
##########
server/blob/blob-postgres/src/test/java/org/apache/james/blob/postgres/PostgresBlobStoreGCAlgorithmTest.java:
##########
@@ -50,6 +49,7 @@ public class PostgresBlobStoreGCAlgorithmTest implements BloomFilterGCAlgorithmC
@BeforeEach
void beforeEach() {
+ System.setProperty("james.postgresql.query.batch.size", "10");
Review Comment:
Do not forget to unset it at the end of the test
##########
mailbox/postgres/src/main/java/org/apache/james/mailbox/postgres/mail/dao/PostgresMailboxMessageDAO.java:
##########
@@ -271,51 +271,62 @@ public Mono<Pair<Integer, Integer>> countTotalAndUnseenMessagesByMailboxId(Postg
public Flux<Pair<SimpleMailboxMessage.Builder, Record>> findMessagesByMailboxId(PostgresMailboxId mailboxId, Limit limit, MessageMapper.FetchType fetchType) {
if (limit.isUnlimited()) {
- return Flux.range(0, OFFSET_UNLIMITED)
- .concatMap(offsetIndex -> findMessagesByMailboxIdBatch(mailboxId, fetchType, offsetIndex * QUERY_BATCH_SIZE, QUERY_BATCH_SIZE))
- .takeUntil(List::isEmpty)
+ return Flux.defer(() -> findMessagesByMailboxIdBatch(mailboxId, fetchType, Optional.empty(), QUERY_BATCH_SIZE))
+ .expand(messages -> {
Review Comment:
<3
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1856047971
##########
server/blob/blob-postgres/src/test/java/org/apache/james/blob/postgres/PostgresBlobStoreGCAlgorithmTest.java:
##########
@@ -0,0 +1,44 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ ****************************************************************/
+
+package org.apache.james.blob.postgres;
+
+import org.apache.james.backends.postgres.PostgresExtension;
+import org.apache.james.blob.api.BlobStoreDAO;
+import org.apache.james.blob.api.PlainBlobId;
+import org.apache.james.server.blob.deduplication.BloomFilterGCAlgorithmContract;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+public class PostgresBlobStoreGCAlgorithmTest implements BloomFilterGCAlgorithmContract {
+
+ @RegisterExtension
+ static PostgresExtension postgresExtension = PostgresExtension.withoutRowLevelSecurity(PostgresBlobStorageModule.MODULE, PostgresExtension.PoolSize.LARGE);
+ private PostgresBlobStoreDAO blobStore;
+
+ @BeforeEach
+ void beforeEach() {
+ blobStore = new PostgresBlobStoreDAO(postgresExtension.getDefaultPostgresExecutor(), new PlainBlobId.Factory(), 1000);
+ }
+
+ @Override
+ public BlobStoreDAO blobStoreDAO() {
+ return blobStore;
+ }
Review Comment:
Do a GC test when deleting 5000+ blobs for me to buy it!
##########
mailbox/postgres/src/main/java/org/apache/james/mailbox/postgres/mail/dao/PostgresMailboxDAO.java:
##########
@@ -202,11 +204,18 @@ public Flux<PostgresMailbox> findMailboxByIds(List<PostgresMailboxId> mailboxIds
public Flux<PostgresMailbox> findMailboxWithPathLike(MailboxQuery.UserBound query) {
String pathLike = MailboxExpressionBackwardCompatibility.getPathLike(query);
+ Function<MailboxQuery.UserBound, Condition> getQueryCondition = mailboxQuery -> {
+ Condition baseCondition = USER_NAME.eq(mailboxQuery.getFixedUser().asString())
+ .and(MAILBOX_NAMESPACE.eq(mailboxQuery.getFixedNamespace()));
+
+ if (Wildcard.INSTANCE.equals(mailboxQuery.getMailboxNameExpression())) {
+ return baseCondition;
+ }
+ return baseCondition.and(MAILBOX_NAME.like(pathLike));
+ };
return postgresExecutor.executeRows(dsl -> Flux.from(dsl.selectFrom(TABLE_NAME)
- .where(MAILBOX_NAME.like(pathLike)
Review Comment:
Fetch eagerly and remove the .collectList.flatmapIterable below?
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2490279815
Any way we might get something prod worthy with a big enough connection pool I bet.
Good news still, though underlying issue might need investigations.
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 [james-project]
Posted by "Arsnael (via GitHub)" <gi...@apache.org>.
Arsnael commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2490290381
Hmmmm... related? ^^'
https://ci-builds.apache.org/job/james/job/ApacheJames/job/PR-2514/3/testReport/org.apache.james.mailbox.postgres.mail.task/PostgresRecomputeCurrentQuotasServiceTest/recomputeCurrentQuotasShouldUpdateContextWhenIncorrectQuotas/
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2500027020
> I noticed that the overall performance did not change significantly.
Almost halth mean time, more than halth for requests relying solely on PGSQL, the results are actually CRAZY good.
This tends to demonstrate a lower contention onto the PG query execution. The metric you provided shows a `x10` drop in tail latency (note that mean latency is unchanged). THis drop in tail latency in the execution of just a few requests fully explains the overall good gains.
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1853445059
##########
backends-common/postgres/src/main/java/org/apache/james/backends/postgres/utils/PostgresExecutor.java:
##########
@@ -125,16 +126,26 @@ public Mono<Void> executeVoid(Function<DSLContext, Mono<?>> queryFunction) {
}
public Flux<Record> executeRows(Function<DSLContext, Flux<Record>> queryFunction) {
+ return executeRows(queryFunction, !EAGER_FETCH);
+ }
+
+ public Flux<Record> executeRows(Function<DSLContext, Flux<Record>> queryFunction, boolean isEagerFetch) {
return Flux.from(metricFactory.decoratePublisherWithTimerMetric("postgres-execution",
Flux.usingWhen(getConnection(domain),
- connection -> dslContext(connection)
- .flatMapMany(queryFunction)
- .timeout(postgresConfiguration.getJooqReactiveTimeout())
- .doOnError(TimeoutException.class, e -> LOGGER.error(JOOQ_TIMEOUT_ERROR_LOG, e))
- .collectList()
- .flatMapIterable(list -> list) // Mitigation fix for https://github.com/jOOQ/jOOQ/issues/16556
- .retryWhen(Retry.backoff(MAX_RETRY_ATTEMPTS, MIN_BACKOFF)
- .filter(preparedStatementConflictException())),
+ connection -> {
+ Flux<Record> recordFlux = dslContext(connection)
+ .flatMapMany(queryFunction)
+ .timeout(postgresConfiguration.getJooqReactiveTimeout())
+ .doOnError(TimeoutException.class, e -> LOGGER.error(JOOQ_TIMEOUT_ERROR_LOG, e))
+ .retryWhen(Retry.backoff(MAX_RETRY_ATTEMPTS, MIN_BACKOFF)
+ .filter(preparedStatementConflictException()));
+
+ if (isEagerFetch) {
Review Comment:
Link an issue in a comment to explain this.
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1853446469
##########
backends-common/postgres/src/main/java/org/apache/james/backends/postgres/utils/PostgresExecutor.java:
##########
@@ -125,16 +126,26 @@ public Mono<Void> executeVoid(Function<DSLContext, Mono<?>> queryFunction) {
}
public Flux<Record> executeRows(Function<DSLContext, Flux<Record>> queryFunction) {
+ return executeRows(queryFunction, !EAGER_FETCH);
+ }
+
+ public Flux<Record> executeRows(Function<DSLContext, Flux<Record>> queryFunction, boolean isEagerFetch) {
Review Comment:
Document in a comment what `isEagerFetch` is doing, why it is needed, when it is Ok to use, when it is not Ok to use, and which are the alternative I can use.
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "vttranlina (via GitHub)" <gi...@apache.org>.
vttranlina commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2493056820
I tried to research `Cursor` , but it does not resolve this issue, the connection still "lock" when cursor in processing.
Then pagination with `limit` + `offset` is last option.
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "vttranlina (via GitHub)" <gi...@apache.org>.
vttranlina commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2499627018
## Performance test result
I noticed that the overall performance did not change significantly.
Some results even showed improvements, but in my opinion, the differences are not very clear.
(likely due to from infrastructor the test environment leading to variations in results across repeated runs of the same test scenario.)
Detail:
### Gatling result
- with IMAP, 10k user, Duration= 30 minutes
### Metric
- Before
```
postgres_execution{quantile="0.5",} 0.001622015
postgres_execution{quantile="0.75",} 0.0028016630000000002
postgres_execution{quantile="0.95",} 0.021495807000000002
postgres_execution{quantile="0.98",} 0.07864319900000001
postgres_execution{quantile="0.99",} 0.149946367
postgres_execution{quantile="0.999",} 0.599785471
postgres_execution_count 3355670.0
```
- After
```
postgres_execution{quantile="0.5",} 0.0017530870000000002
postgres_execution{quantile="0.75",} 0.0025722870000000004
postgres_execution{quantile="0.95",} 0.005931007
postgres_execution{quantile="0.98",} 0.009895935
postgres_execution{quantile="0.99",} 0.016056319
postgres_execution{quantile="0.999",} 0.057147391000000006
postgres_execution_count 2591175.0
```
### pg_stat_statement
1. Focus to pagination query
**A. Before**
```sql
select * from message_mailbox where( mailbox_id = cast($1 as uuid) and is_recent = $2) order by message_uid asc
```
- mean_exec_time = 0.36819846626378827
- total_exec_time = 18826.72397700015
- calls = 51132
```sql
select * from message_mailbox where( mailbox_id = cast($1 as uuid) and message_uid >= $2 and message_uid <= $3) order by message_uid asc
```
- mean_exec_time = 0.08792518239753881
- total_exec_time = 2400.621255000017
- calls = 27303
**B. After**
```sql
select * from message_mailbox where( mailbox_id = cast($1 as uuid) and is_recent = $2) order by message_uid asc offset $3 rows fetch next $4 rows only
```
- mean_exec_time = 0.12503941170118069
- total_exec_time = 6599.705188999946
- calls = 52781
```sql
select * from message_mailbox where( mailbox_id = cast($1 as uuid) and message_uid >= $2 and message_uid <= $3) order by message_uid asc offset $4 rows fetch next $5 rows only
```
- mean_exec_time = 0.044664966889503485
- total_exec_time = 2455379999974
- calls = 57142
3. Overview
- Before
- After
// File Result Details: https://github.com/vttranlina/vttranlina.github.io/tree/main/gatling_20241126
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "vttranlina (via GitHub)" <gi...@apache.org>.
vttranlina commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1853409698
##########
mailbox/postgres/src/main/java/org/apache/james/mailbox/postgres/mail/dao/PostgresMailboxMessageDAO.java:
##########
@@ -117,78 +116,107 @@ 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;
public PostgresMailboxMessageDAO(PostgresExecutor postgresExecutor) {
this.postgresExecutor = postgresExecutor;
}
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 * BATCH_SIZE, BATCH_SIZE))
+ .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)
+ .from(TABLE_NAME)
+ .where(MAILBOX_ID.eq((mailboxId.asUuid())))
+ .and(IS_SEEN.eq(false))
+ .orderBy(DEFAULT_SORT_ORDER_BY)
+ .limit(batchSize)
+ .offset(offset)))
+ .map(RECORD_TO_MESSAGE_UID_FUNCTION)
+ .collectList()
Review Comment:
Note: we still need `collectionList` in small block.
If not, the issue https://github.com/apache/james-project/pull/2514#issuecomment-2490210967 still happend.
TLDR: if we don't want to use `collectionList`
the batchSize must be < `32` when reactor configuration by default (IMO, 32 is quiet small for batch size)
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1853444444
##########
backends-common/postgres/src/main/java/org/apache/james/backends/postgres/PostgresCommons.java:
##########
@@ -93,4 +93,8 @@ public static <T> Field<T> tableField(Table<Record> table, Field<T> field) {
public static final int IN_CLAUSE_MAX_SIZE = 32;
+ public static final int BATCH_SIZE = 100;
Review Comment:
Make this configurable?
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 [james-project]
Posted by "vttranlina (via GitHub)" <gi...@apache.org>.
vttranlina commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2490412932
> Hmmmm... related? ^^'
>
> https://ci-builds.apache.org/job/james/job/ApacheJames/job/PR-2514/3/testReport/org.apache.james.mailbox.postgres.mail.task/PostgresRecomputeCurrentQuotasServiceTest/recomputeCurrentQuotasShouldUpdateContextWhenIncorrectQuotas/
yes, that I posted at https://github.com/apache/james-project/pull/2514#issuecomment-2490210967
The test using `PoolSize.SMALL(1, 2)`, that is reason
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1855134539
##########
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:
No Really for UID I strongly believe .collectList would be better.
> the batch help we early release connection to pool
Wrong assumption: yes this wont lock one connection the entire time but we'd have many request: usage would be higher.
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "vttranlina (via GitHub)" <gi...@apache.org>.
vttranlina commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2496949611
> Let's look for bugs: can we mark 1000 messages as deleted and then expunge and ensure we did delete then all?
I added the test case `expungeMessageShouldCorrectWhenALotOfMessages`. It always works because `retrieveMessagesMarkedForDeletion` returns collectionList,
And we use `.collectionList` at DAO layer
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1858161149
##########
mailbox/postgres/src/main/java/org/apache/james/mailbox/postgres/mail/dao/PostgresMailboxMessageDAO.java:
##########
@@ -273,31 +270,55 @@ public Mono<Pair<Integer, Integer>> countTotalAndUnseenMessagesByMailboxId(Postg
}
public Flux<Pair<SimpleMailboxMessage.Builder, Record>> findMessagesByMailboxId(PostgresMailboxId mailboxId, Limit limit, MessageMapper.FetchType fetchType) {
- PostgresMailboxMessageFetchStrategy fetchStrategy = FETCH_TYPE_TO_FETCH_STRATEGY.apply(fetchType);
- Function<DSLContext, SelectSeekStep1<Record, Long>> queryWithoutLimit = dslContext -> dslContext.select(fetchStrategy.fetchFields())
- .from(MESSAGES_JOIN_MAILBOX_MESSAGES_CONDITION_STEP)
- .where(MAILBOX_ID.eq(mailboxId.asUuid()))
- .orderBy(DEFAULT_SORT_ORDER_BY);
+ if (limit.isUnlimited()) {
+ return Flux.range(0, OFFSET_UNLIMITED)
+ .concatMap(offsetIndex -> findMessagesByMailboxIdBatch(mailboxId, fetchType, offsetIndex * queryBatchSize, queryBatchSize))
+ .takeUntil(List::isEmpty)
+ .flatMapIterable(Function.identity());
+ } else {
+ return findMessagesByMailboxIdBatch(mailboxId, fetchType, 0, limit.getLimit().get())
+ .flatMapIterable(Function.identity());
+ }
+ }
- return postgresExecutor.executeRows(dslContext -> limit.getLimit()
- .map(limitValue -> Flux.from(queryWithoutLimit.andThen(step -> step.limit(limitValue)).apply(dslContext)))
- .orElse(Flux.from(queryWithoutLimit.apply(dslContext))))
- .map(record -> Pair.of(fetchStrategy.toMessageBuilder().apply(record), record));
+ private Mono<List<Pair<SimpleMailboxMessage.Builder, Record>>> findMessagesByMailboxIdBatch(PostgresMailboxId mailboxId, MessageMapper.FetchType fetchType, int offset, int batchSize) {
+ PostgresMailboxMessageFetchStrategy fetchStrategy = FETCH_TYPE_TO_FETCH_STRATEGY.apply(fetchType);
+ return postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.select(fetchStrategy.fetchFields())
+ .from(MESSAGES_JOIN_MAILBOX_MESSAGES_CONDITION_STEP)
+ .where(MAILBOX_ID.eq(mailboxId.asUuid()))
+ .orderBy(DEFAULT_SORT_ORDER_BY)
+ .limit(batchSize)
+ .offset(offset)))
Review Comment:
> Should we also implement this for the Postgres BlobStore DAO?
It would rely on alphabetic ordering, it is likely less important.
##########
mailbox/postgres/src/main/java/org/apache/james/mailbox/postgres/mail/dao/PostgresMailboxMessageDAO.java:
##########
@@ -273,31 +270,55 @@ public Mono<Pair<Integer, Integer>> countTotalAndUnseenMessagesByMailboxId(Postg
}
public Flux<Pair<SimpleMailboxMessage.Builder, Record>> findMessagesByMailboxId(PostgresMailboxId mailboxId, Limit limit, MessageMapper.FetchType fetchType) {
- PostgresMailboxMessageFetchStrategy fetchStrategy = FETCH_TYPE_TO_FETCH_STRATEGY.apply(fetchType);
- Function<DSLContext, SelectSeekStep1<Record, Long>> queryWithoutLimit = dslContext -> dslContext.select(fetchStrategy.fetchFields())
- .from(MESSAGES_JOIN_MAILBOX_MESSAGES_CONDITION_STEP)
- .where(MAILBOX_ID.eq(mailboxId.asUuid()))
- .orderBy(DEFAULT_SORT_ORDER_BY);
+ if (limit.isUnlimited()) {
+ return Flux.range(0, OFFSET_UNLIMITED)
+ .concatMap(offsetIndex -> findMessagesByMailboxIdBatch(mailboxId, fetchType, offsetIndex * queryBatchSize, queryBatchSize))
+ .takeUntil(List::isEmpty)
+ .flatMapIterable(Function.identity());
+ } else {
+ return findMessagesByMailboxIdBatch(mailboxId, fetchType, 0, limit.getLimit().get())
+ .flatMapIterable(Function.identity());
+ }
+ }
- return postgresExecutor.executeRows(dslContext -> limit.getLimit()
- .map(limitValue -> Flux.from(queryWithoutLimit.andThen(step -> step.limit(limitValue)).apply(dslContext)))
- .orElse(Flux.from(queryWithoutLimit.apply(dslContext))))
- .map(record -> Pair.of(fetchStrategy.toMessageBuilder().apply(record), record));
+ private Mono<List<Pair<SimpleMailboxMessage.Builder, Record>>> findMessagesByMailboxIdBatch(PostgresMailboxId mailboxId, MessageMapper.FetchType fetchType, int offset, int batchSize) {
+ PostgresMailboxMessageFetchStrategy fetchStrategy = FETCH_TYPE_TO_FETCH_STRATEGY.apply(fetchType);
+ return postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.select(fetchStrategy.fetchFields())
+ .from(MESSAGES_JOIN_MAILBOX_MESSAGES_CONDITION_STEP)
+ .where(MAILBOX_ID.eq(mailboxId.asUuid()))
+ .orderBy(DEFAULT_SORT_ORDER_BY)
+ .limit(batchSize)
+ .offset(offset)))
Review Comment:
> Should we also implement this for the Postgres BlobStore DAO?
It would rely on alphabetic ordering, it is likely less performant.
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1858151717
##########
mailbox/postgres/src/main/java/org/apache/james/mailbox/postgres/mail/dao/PostgresMailboxMessageDAO.java:
##########
@@ -273,31 +270,55 @@ public Mono<Pair<Integer, Integer>> countTotalAndUnseenMessagesByMailboxId(Postg
}
public Flux<Pair<SimpleMailboxMessage.Builder, Record>> findMessagesByMailboxId(PostgresMailboxId mailboxId, Limit limit, MessageMapper.FetchType fetchType) {
- PostgresMailboxMessageFetchStrategy fetchStrategy = FETCH_TYPE_TO_FETCH_STRATEGY.apply(fetchType);
- Function<DSLContext, SelectSeekStep1<Record, Long>> queryWithoutLimit = dslContext -> dslContext.select(fetchStrategy.fetchFields())
- .from(MESSAGES_JOIN_MAILBOX_MESSAGES_CONDITION_STEP)
- .where(MAILBOX_ID.eq(mailboxId.asUuid()))
- .orderBy(DEFAULT_SORT_ORDER_BY);
+ if (limit.isUnlimited()) {
+ return Flux.range(0, OFFSET_UNLIMITED)
+ .concatMap(offsetIndex -> findMessagesByMailboxIdBatch(mailboxId, fetchType, offsetIndex * queryBatchSize, queryBatchSize))
+ .takeUntil(List::isEmpty)
+ .flatMapIterable(Function.identity());
+ } else {
+ return findMessagesByMailboxIdBatch(mailboxId, fetchType, 0, limit.getLimit().get())
+ .flatMapIterable(Function.identity());
+ }
+ }
- return postgresExecutor.executeRows(dslContext -> limit.getLimit()
- .map(limitValue -> Flux.from(queryWithoutLimit.andThen(step -> step.limit(limitValue)).apply(dslContext)))
- .orElse(Flux.from(queryWithoutLimit.apply(dslContext))))
- .map(record -> Pair.of(fetchStrategy.toMessageBuilder().apply(record), record));
+ private Mono<List<Pair<SimpleMailboxMessage.Builder, Record>>> findMessagesByMailboxIdBatch(PostgresMailboxId mailboxId, MessageMapper.FetchType fetchType, int offset, int batchSize) {
+ PostgresMailboxMessageFetchStrategy fetchStrategy = FETCH_TYPE_TO_FETCH_STRATEGY.apply(fetchType);
+ return postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.select(fetchStrategy.fetchFields())
+ .from(MESSAGES_JOIN_MAILBOX_MESSAGES_CONDITION_STEP)
+ .where(MAILBOX_ID.eq(mailboxId.asUuid()))
+ .orderBy(DEFAULT_SORT_ORDER_BY)
+ .limit(batchSize)
+ .offset(offset)))
Review Comment:
Actually we would be WAY more resilient not using an offset. We SHALL reuse the highest UID from the previous batch and resume from it. This would make us reslient to deletion in already paged items.
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 & resolve hanging issue [james-project]
Posted by "vttranlina (via GitHub)" <gi...@apache.org>.
vttranlina commented on code in PR #2514:
URL: https://github.com/apache/james-project/pull/2514#discussion_r1858157535
##########
mailbox/postgres/src/main/java/org/apache/james/mailbox/postgres/mail/dao/PostgresMailboxMessageDAO.java:
##########
@@ -273,31 +270,55 @@ public Mono<Pair<Integer, Integer>> countTotalAndUnseenMessagesByMailboxId(Postg
}
public Flux<Pair<SimpleMailboxMessage.Builder, Record>> findMessagesByMailboxId(PostgresMailboxId mailboxId, Limit limit, MessageMapper.FetchType fetchType) {
- PostgresMailboxMessageFetchStrategy fetchStrategy = FETCH_TYPE_TO_FETCH_STRATEGY.apply(fetchType);
- Function<DSLContext, SelectSeekStep1<Record, Long>> queryWithoutLimit = dslContext -> dslContext.select(fetchStrategy.fetchFields())
- .from(MESSAGES_JOIN_MAILBOX_MESSAGES_CONDITION_STEP)
- .where(MAILBOX_ID.eq(mailboxId.asUuid()))
- .orderBy(DEFAULT_SORT_ORDER_BY);
+ if (limit.isUnlimited()) {
+ return Flux.range(0, OFFSET_UNLIMITED)
+ .concatMap(offsetIndex -> findMessagesByMailboxIdBatch(mailboxId, fetchType, offsetIndex * queryBatchSize, queryBatchSize))
+ .takeUntil(List::isEmpty)
+ .flatMapIterable(Function.identity());
+ } else {
+ return findMessagesByMailboxIdBatch(mailboxId, fetchType, 0, limit.getLimit().get())
+ .flatMapIterable(Function.identity());
+ }
+ }
- return postgresExecutor.executeRows(dslContext -> limit.getLimit()
- .map(limitValue -> Flux.from(queryWithoutLimit.andThen(step -> step.limit(limitValue)).apply(dslContext)))
- .orElse(Flux.from(queryWithoutLimit.apply(dslContext))))
- .map(record -> Pair.of(fetchStrategy.toMessageBuilder().apply(record), record));
+ private Mono<List<Pair<SimpleMailboxMessage.Builder, Record>>> findMessagesByMailboxIdBatch(PostgresMailboxId mailboxId, MessageMapper.FetchType fetchType, int offset, int batchSize) {
+ PostgresMailboxMessageFetchStrategy fetchStrategy = FETCH_TYPE_TO_FETCH_STRATEGY.apply(fetchType);
+ return postgresExecutor.executeRows(dslContext -> Flux.from(dslContext.select(fetchStrategy.fetchFields())
+ .from(MESSAGES_JOIN_MAILBOX_MESSAGES_CONDITION_STEP)
+ .where(MAILBOX_ID.eq(mailboxId.asUuid()))
+ .orderBy(DEFAULT_SORT_ORDER_BY)
+ .limit(batchSize)
+ .offset(offset)))
Review Comment:
Good idea
Should we also implement this for the Postgres BlobStore DAO?
We can add an additional column, such as a sequence or counter column?
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
Re: [PR] JAMES-2586 - Bump Jooq 3.19.13 -> 3.19.15 [james-project]
Posted by "chibenwa (via GitHub)" <gi...@apache.org>.
chibenwa commented on PR #2514:
URL: https://github.com/apache/james-project/pull/2514#issuecomment-2489044848
Did we just kill it?
Can I open champagne tonight?
--
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-unsubscribe@james.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org