This is an automated email from the ASF dual-hosted git repository.
chibenwa pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
The following commit(s) were added to refs/heads/master by this push:
new 73cb2e9dae [FIX] PGSQL: always init HSTORE correctly
73cb2e9dae is described below
commit 73cb2e9daef2626095fb41aad2a9409f8eb03f14
Author: Benoit TELLIER <[email protected]>
AuthorDate: Mon Sep 14 10:49:36 2026 +0200
[FIX] PGSQL: always init HSTORE correctly
---
.../backends/postgres/PostgresTableManager.java | 18 ++-
.../PostgresTableManagerHstoreExtensionTest.java | 135 +++++++++++++++++++++
.../james/modules/data/PostgresCommonModule.java | 3 +-
3 files changed, 151 insertions(+), 5 deletions(-)
diff --git
a/backends-common/postgres/src/main/java/org/apache/james/backends/postgres/PostgresTableManager.java
b/backends-common/postgres/src/main/java/org/apache/james/backends/postgres/PostgresTableManager.java
index a3140dfa91..1e2fc5f533 100644
---
a/backends-common/postgres/src/main/java/org/apache/james/backends/postgres/PostgresTableManager.java
+++
b/backends-common/postgres/src/main/java/org/apache/james/backends/postgres/PostgresTableManager.java
@@ -20,6 +20,7 @@
package org.apache.james.backends.postgres;
import java.util.List;
+import java.util.function.Supplier;
import jakarta.inject.Inject;
@@ -28,12 +29,14 @@ import org.apache.james.lifecycle.api.Startable;
import org.jooq.DSLContext;
import org.jooq.exception.DataAccessException;
import org.jooq.impl.DSL;
+import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.annotations.VisibleForTesting;
import io.r2dbc.spi.Connection;
+import io.r2dbc.spi.ConnectionFactory;
import io.r2dbc.spi.Result;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -42,14 +45,21 @@ public class PostgresTableManager implements Startable {
public static final int INITIALIZATION_PRIORITY = 1;
private static final Logger LOGGER =
LoggerFactory.getLogger(PostgresTableManager.class);
private final PostgresExecutor postgresExecutor;
+ private final Supplier<Publisher<? extends Connection>>
extensionConnectionSupplier;
private final PostgresDataDefinition module;
private final RowLevelSecurity rowLevelSecurity;
+ /**
+ * @param connectionFactory the non pooled factory the pool of {@code
postgresExecutor} is backed by. The hstore
+ * extension is created with it, see {@link
#initializePostgresExtension()}.
+ */
@Inject
public PostgresTableManager(PostgresExecutor postgresExecutor,
+ ConnectionFactory connectionFactory,
PostgresDataDefinition module,
PostgresConfiguration postgresConfiguration) {
this.postgresExecutor = postgresExecutor;
+ this.extensionConnectionSupplier = connectionFactory::create;
this.module = module;
this.rowLevelSecurity = postgresConfiguration.getRowLevelSecurity();
}
@@ -57,6 +67,7 @@ public class PostgresTableManager implements Startable {
@VisibleForTesting
public PostgresTableManager(PostgresExecutor postgresExecutor,
PostgresDataDefinition module, RowLevelSecurity rowLevelSecurity) {
this.postgresExecutor = postgresExecutor;
+ this.extensionConnectionSupplier = () ->
postgresExecutor.connectionFactory().getConnection();
this.module = module;
this.rowLevelSecurity = rowLevelSecurity;
}
@@ -69,13 +80,12 @@ public class PostgresTableManager implements Startable {
}
public Mono<Void> initializePostgresExtension() {
- return
Mono.usingWhen(postgresExecutor.connectionFactory().getConnection(),
- connection -> Mono.just(connection)
- .flatMapMany(pgConnection ->
pgConnection.createStatement("CREATE EXTENSION IF NOT EXISTS hstore")
+ return Mono.usingWhen(extensionConnectionSupplier.get(), // Not
pooled: r2dbc only registers the hstore codec on connections opened after the
extension exists
+ connection -> Flux.from(connection.createStatement("CREATE
EXTENSION IF NOT EXISTS hstore")
.execute())
.flatMap(Result::getRowsUpdated)
.then(),
- connection ->
postgresExecutor.connectionFactory().closeConnection(connection));
+ Connection::close);
}
public Mono<Void> initializeTables() {
diff --git
a/backends-common/postgres/src/test/java/org/apache/james/backends/postgres/PostgresTableManagerHstoreExtensionTest.java
b/backends-common/postgres/src/test/java/org/apache/james/backends/postgres/PostgresTableManagerHstoreExtensionTest.java
new file mode 100644
index 0000000000..c56e767df4
--- /dev/null
+++
b/backends-common/postgres/src/test/java/org/apache/james/backends/postgres/PostgresTableManagerHstoreExtensionTest.java
@@ -0,0 +1,135 @@
+/****************************************************************
+ * 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.backends.postgres;
+
+import static
org.apache.james.backends.postgres.PostgresFixture.Database.DEFAULT_DATABASE;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.time.Duration;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+
+import
org.apache.james.backends.postgres.utils.PoolBackedPostgresConnectionFactory;
+import org.apache.james.backends.postgres.utils.PostgresExecutor;
+import org.apache.james.metrics.tests.RecordingMetricFactory;
+import org.jooq.Field;
+import org.jooq.SQLDialect;
+import org.jooq.Table;
+import org.jooq.impl.DSL;
+import org.jooq.postgres.extensions.types.Hstore;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.PostgreSQLContainer;
+
+import io.r2dbc.postgresql.PostgresqlConnectionConfiguration;
+import io.r2dbc.postgresql.PostgresqlConnectionFactory;
+import io.r2dbc.spi.Connection;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+class PostgresTableManagerHstoreExtensionTest {
+ private static final int POOL_SIZE = 3;
+ private static final Table<?> TABLE = DSL.table("hstore_table");
+ private static final Field<Hstore> HSTORE_COLUMN =
DSL.field("hstore_column", PostgresCommons.DataTypes.HSTORE);
+ private static final PostgresDataDefinition MODULE =
PostgresDataDefinition.table(PostgresTable.name(TABLE.getName())
+ .createTableStep((dsl, tableName) ->
dsl.createTableIfNotExists(tableName)
+ .column(HSTORE_COLUMN))
+ .disableRowLevelSecurity()
+ .build());
+
+ private final PostgreSQLContainer<?> container =
DockerPostgresSingleton.SINGLETON;
+ private String databaseName;
+ private PoolBackedPostgresConnectionFactory pool;
+ private PostgresExecutor postgresExecutor;
+ private PostgresqlConnectionFactory connectionFactory;
+ private PostgresConfiguration postgresConfiguration;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ // A database of its own: the one of PostgresExtension already has the
hstore extension
+ databaseName = "hstore_" + UUID.randomUUID().toString().replace("-",
"");
+ container.execInContainer("psql", "-U", DEFAULT_DATABASE.dbUser(),
"-c", "CREATE DATABASE " + databaseName + ";");
+
+ postgresConfiguration = PostgresConfiguration.builder()
+ .databaseName(databaseName)
+ .databaseSchema(DEFAULT_DATABASE.schema())
+ .host(container.getHost())
+ .port(container.getMappedPort(PostgresFixture.PORT))
+ .username(DEFAULT_DATABASE.dbUser())
+ .password(DEFAULT_DATABASE.dbPassword())
+ .rowLevelSecurityEnabled(false)
+ .jooqReactiveTimeout(Optional.of(Duration.ofSeconds(20L)))
+ .build();
+ connectionFactory = new
PostgresqlConnectionFactory(PostgresqlConnectionConfiguration.builder()
+ .host(postgresConfiguration.getHost())
+ .port(postgresConfiguration.getPort())
+ .database(databaseName)
+ .schema(DEFAULT_DATABASE.schema())
+ .username(DEFAULT_DATABASE.dbUser())
+ .password(DEFAULT_DATABASE.dbPassword())
+ .build());
+ pool = new
PoolBackedPostgresConnectionFactory(RowLevelSecurity.DISABLED, POOL_SIZE,
POOL_SIZE, connectionFactory);
+ postgresExecutor = new PostgresExecutor.Factory(pool,
postgresConfiguration, new RecordingMetricFactory()).create();
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ pool.close().block();
+ container.execInContainer("psql", "-U", DEFAULT_DATABASE.dbUser(),
"-c", "DROP DATABASE IF EXISTS " + databaseName + " WITH (FORCE);");
+ }
+
+ @Test
+ void
everyPooledConnectionShouldDecodeHstoreWhenInitializingAFreshDatabase() {
+ new PostgresTableManager(postgresExecutor, connectionFactory, MODULE,
postgresConfiguration)
+ .initPostgres();
+
+ Map<String, String> entries = Map.of("[email protected]", "lr",
"[email protected]", "aeiklprstwx");
+ postgresExecutor.executeVoid(dsl -> Mono.from(dsl.insertInto(TABLE,
HSTORE_COLUMN)
+ .values(Hstore.hstore(entries))))
+ .block();
+
+ assertThat(readHstoreOnEveryPooledConnection())
+ .hasSize(POOL_SIZE)
+ .allSatisfy(read -> assertThat(read).isEqualTo(entries));
+ }
+
+ private List<Map<?, ?>> readHstoreOnEveryPooledConnection() {
+ return Flux.range(0, POOL_SIZE)
+ .flatMap(any -> pool.getConnection())
+ .collectList()
+ .flatMapMany(connections -> Flux.fromIterable(connections)
+ .concatMap(this::readHstore)
+ .concatWith(Flux.fromIterable(connections)
+ .concatMap(connection -> pool.closeConnection(connection))
+ .then(Mono.empty())))
+ .collectList()
+ .block();
+ }
+
+ private Mono<Map<?, ?>> readHstore(Connection connection) {
+ // An untyped select, as the DAOs do (selectFrom(TABLE_NAME) then
record.get(field, LinkedHashMap.class))
+ return Mono.from(DSL.using(connection,
SQLDialect.POSTGRES).selectFrom(TABLE))
+ .map(record -> record.get(HSTORE_COLUMN, LinkedHashMap.class));
+ }
+}
diff --git
a/server/container/guice/postgres-common/src/main/java/org/apache/james/modules/data/PostgresCommonModule.java
b/server/container/guice/postgres-common/src/main/java/org/apache/james/modules/data/PostgresCommonModule.java
index 05fa938f92..d8216a0352 100644
---
a/server/container/guice/postgres-common/src/main/java/org/apache/james/modules/data/PostgresCommonModule.java
+++
b/server/container/guice/postgres-common/src/main/java/org/apache/james/modules/data/PostgresCommonModule.java
@@ -137,9 +137,10 @@ public class PostgresCommonModule extends AbstractModule {
@Provides
@Singleton
PostgresTableManager postgresTableManager(PostgresExecutor
postgresExecutor,
+ ConnectionFactory
connectionFactory,
PostgresDataDefinition
postgresDataDefinition,
PostgresConfiguration
postgresConfiguration) {
- return new PostgresTableManager(postgresExecutor,
postgresDataDefinition, postgresConfiguration);
+ return new PostgresTableManager(postgresExecutor, connectionFactory,
postgresDataDefinition, postgresConfiguration);
}
@Provides
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
