diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 165d8e41e..4c31ebc91 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -3,7 +3,6 @@ antlr = '4.13.2' xsom = '20140925' aalto = '1.4.0' staxmate = '2.4.2' -rxjava-jdbc = '0.1.4-ii.1' hikaricp = '7.1.0' postgresql = '42.7.13' sqlite = '3.53.4.0' @@ -28,7 +27,6 @@ aalto = { module = "com.fasterxml:aalto-xml", version.ref = "aalto" } staxmate = { module = "com.fasterxml.staxmate:staxmate", version.ref = "staxmate" } # sql -rxjava-jdbc = { module = "com.github.davidmoten:rxjava3-jdbc", version.ref = "rxjava-jdbc" } hikaricp = { module = "com.zaxxer:HikariCP", version.ref = "hikaricp" } postgresql = { module = "org.postgresql:postgresql", version.ref = "postgresql" } sqlite = { module = "org.xerial:sqlite-jdbc", version.ref = "sqlite" } diff --git a/xtraplatform-features-geoparquet/src/main/java/de/ii/xtraplatform/features/geoparquet/app/SqlDbmsAdapterDuckdb.java b/xtraplatform-features-geoparquet/src/main/java/de/ii/xtraplatform/features/geoparquet/app/SqlDbmsAdapterDuckdb.java index 6eda39cfb..28d2026cd 100644 --- a/xtraplatform-features-geoparquet/src/main/java/de/ii/xtraplatform/features/geoparquet/app/SqlDbmsAdapterDuckdb.java +++ b/xtraplatform-features-geoparquet/src/main/java/de/ii/xtraplatform/features/geoparquet/app/SqlDbmsAdapterDuckdb.java @@ -26,7 +26,6 @@ import java.util.Map; import java.util.Optional; import javax.sql.DataSource; -import org.davidmoten.rxjava3.jdbc.pool.DatabaseType; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -251,11 +250,6 @@ public List getDefaultSchemas() { return List.of(); } - @Override - public DatabaseType getRxType() { - return DatabaseType.OTHER; - } - @Override public List getSystemSchemas() { return List.of("public"); diff --git a/xtraplatform-features-oracle/src/main/java/de/ii/xtraplatform/features/oracle/app/SqlDbmsAdapterOras.java b/xtraplatform-features-oracle/src/main/java/de/ii/xtraplatform/features/oracle/app/SqlDbmsAdapterOras.java index 1fc57de8b..5de9af646 100644 --- a/xtraplatform-features-oracle/src/main/java/de/ii/xtraplatform/features/oracle/app/SqlDbmsAdapterOras.java +++ b/xtraplatform-features-oracle/src/main/java/de/ii/xtraplatform/features/oracle/app/SqlDbmsAdapterOras.java @@ -24,7 +24,6 @@ import java.util.Optional; import java.util.Set; import javax.sql.DataSource; -import org.davidmoten.rxjava3.jdbc.pool.DatabaseType; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -128,11 +127,6 @@ public Optional getInitSql(ConnectionInfoSql connectionInfo) { return Optional.empty(); } - @Override - public DatabaseType getRxType() { - return DatabaseType.ORACLE; - } - @Override public List getDefaultSchemas() { return List.of("public"); diff --git a/xtraplatform-features-sql/build.gradle b/xtraplatform-features-sql/build.gradle index caaee7be7..50c7abe8e 100644 --- a/xtraplatform-features-sql/build.gradle +++ b/xtraplatform-features-sql/build.gradle @@ -23,13 +23,6 @@ dependencies { provided project(':xtraplatform-geometries') provided project(':xtraplatform-strings') - embedded(libs.rxjava.jdbc) { - exclude module: 'rxjava' - exclude module: 'reactive-streams' - exclude module: 'commons-io' - exclude module: 'slf4j-api' - exclude group: 'com.google.code.findbugs' - } //use reactive-streams + rxjava exported from this embeddedImport 'de.interactive_instruments:xtraplatform-streams' diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlDbmsAdapter.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlDbmsAdapter.java index 5eabde8c6..64f6abdff 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlDbmsAdapter.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlDbmsAdapter.java @@ -16,7 +16,6 @@ import java.util.Optional; import javax.annotation.Nullable; import javax.sql.DataSource; -import org.davidmoten.rxjava3.jdbc.pool.DatabaseType; import org.immutables.value.Value; @AutoMultiBind @@ -32,8 +31,6 @@ public interface SqlDbmsAdapter { List getDefaultSchemas(); - DatabaseType getRxType(); - List getSystemSchemas(); List getSystemTables(); diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlClientRx.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlClientRx.java index 7eb721552..9f42b72ef 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlClientRx.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlClientRx.java @@ -20,6 +20,7 @@ import io.reactivex.rxjava3.core.Flowable; import io.reactivex.rxjava3.schedulers.Schedulers; import java.sql.Connection; +import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; import java.text.Collator; @@ -34,7 +35,6 @@ import java.util.stream.Collectors; import java.util.stream.Stream; import javax.sql.DataSource; -import org.davidmoten.rxjava3.jdbc.Database; import org.postgresql.PGConnection; import org.postgresql.PGNotification; import org.slf4j.Logger; @@ -50,21 +50,16 @@ public class SqlClientRx implements SqlClient { // it exists to end. private static final long READ_STALL_TIMEOUT_MINUTES = 10; - // rxjava3-jdbc is used for streamed reads only; everything that needs a connection of its own - // (sessions, statements without a result) leases it from the pool directly - private final Database session; private final DataSource dataSource; private final SqlDbmsAdapter dbmsAdapter; private final SqlDialect dialect; private final Collator collator; public SqlClientRx( - Database session, DataSource dataSource, SqlDbmsAdapter dbmsAdapter, SqlDialect dialect, Optional defaultCollation) { - this.session = session; this.dataSource = dataSource; this.dbmsAdapter = dbmsAdapter; this.dialect = dialect; @@ -92,11 +87,19 @@ public CompletableFuture> run(String query, SqlQueryOptions o return result; } - session - .select(query) - .get(resultSet -> new SqlRowVals(collator).read(resultSet, options)) - .toList() - .subscribe(result::complete, result::completeExceptionally); + try (Connection connection = dataSource.getConnection(); + Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery(query)) { + List rows = new ArrayList<>(); + + while (resultSet.next()) { + rows.add(new SqlRowVals(collator).read(resultSet, options)); + } + + result.complete(rows); + } catch (SQLException | RuntimeException e) { + result.completeExceptionally(e); + } return result; } @@ -108,31 +111,35 @@ public Reactive.Source getSourceStream(String query, SqlQueryOptions opt } List logBuffer = new ArrayList<>(5); - org.davidmoten.rxjava3.jdbc.ResultSetMapper mapper = - resultSet -> { - SqlRow row = new SqlRowVals(collator).read(resultSet, options); - - if (LOGGER.isDebugEnabled(MARKER.SQL_RESULT) && logBuffer.size() < 10) { - logBuffer.add(row); - } - - return row; - }; - // A positive fetch size requires a transaction so the database driver uses a server-side cursor // and streams rows instead of buffering the whole result set in memory (PostgreSQL ignores the // fetch size with autoCommit=true). - // TODO encapsulating the query in a transaction is also a workaround for what appears to be a - // bug in rxjava3-jdbc, see https://github.com/interactive-instruments/ldproxy/issues/1293 + boolean streamed = options.getFetchSize() > 0; + + // The connection is leased when the stream is subscribed and returned to the pool when it + // terminates, whether the rows were exhausted, the read failed or the consumer cancelled. Flowable flowable = - options.getFetchSize() > 0 - ? session - .select(query) - .transacted() - .fetchSize(options.getFetchSize()) - .valuesOnly() - .get(mapper) - : session.select(query).get(mapper); + Flowable.using( + () -> lease(streamed), + connection -> + Flowable.generate( + () -> execute(connection, query, options.getFetchSize()), + (resultSet, emitter) -> { + if (resultSet.next()) { + SqlRow row = new SqlRowVals(collator).read(resultSet, options); + + if (LOGGER.isDebugEnabled(MARKER.SQL_RESULT) && logBuffer.size() < 10) { + logBuffer.add(row); + } + + emitter.onNext(row); + } else { + emitter.onComplete(); + } + }, + SqlClientRx::close), + connection -> release(connection, streamed), + true); // TODO: prettify, see // https://github.com/slick/slick/blob/main/slick/src/main/scala/slick/jdbc/StatementInvoker.scala @@ -169,20 +176,17 @@ public Reactive.Source getSourceStream(String query, SqlQueryOptions opt }); } - // The blocking connection provider runs connect+execute+read on the subscribing thread, so - // without this the whole stream is single-threaded. Subscribing on a worker thread lets several - // parallel-flagged queries (e.g. the concurrent single-shot value phase) run at once, each on - // its - // own connection. + // Lease, execute and read run on the subscribing thread, so without this the whole stream is + // single-threaded. Subscribing on a worker thread lets several parallel-flagged queries (e.g. + // the concurrent single-shot value phase) run at once, each on its own connection. if (options.isParallel()) { flowable = flowable.subscribeOn(Schedulers.io()); } - // Safety net for a read that neither completes nor fails. A database error raised while the - // rows are being streamed is not delivered by the underlying library (see the issue linked - // above), so the stream can stall forever: no error is logged, no response is sent, and the - // connections the sub-query holds stay held until the client gives up. A connection lost - // mid-stream — a failover in a replicated cluster, for instance — looks exactly the same. + // Safety net for a read that neither completes nor fails: a connection lost mid-stream — a + // failover in a replicated cluster, for instance — can leave the driver waiting for the next + // row forever, so no error is logged, no response is sent, and the connections the sub-query + // holds stay held until the client gives up. // The timeout is per element, not per stream, so a slow but progressing read is unaffected // however long it runs in total; only a gap longer than the window ends the stream, with an // error that does propagate. A database-side statement_timeout is no substitute: its error @@ -203,6 +207,80 @@ public Reactive.Source getSourceStream(String query, SqlQueryOptions opt return Reactive.Source.publisher(flowable); } + private Connection lease(boolean transaction) throws SQLException { + Connection connection = dataSource.getConnection(); + + if (transaction) { + try { + connection.setAutoCommit(false); + } catch (SQLException e) { + close(connection); + throw e; + } + } + + return connection; + } + + private static ResultSet execute(Connection connection, String query, int fetchSize) + throws SQLException { + Statement statement = connection.createStatement(); + + try { + if (fetchSize > 0) { + statement.setFetchSize(fetchSize); + } + + return statement.executeQuery(query); + } catch (SQLException | RuntimeException e) { + close(statement); + throw e; + } + } + + /** Ends a read-only transaction, if any, and returns the connection to the pool. */ + private static void release(Connection connection, boolean transaction) { + if (transaction) { + try { + // nothing to commit, and a rollback is the cheapest way to close the server-side cursor + connection.rollback(); + } catch (SQLException e) { + LOGGER.debug("Ending the read transaction failed: {}", e.getMessage()); + } + try { + connection.setAutoCommit(true); + } catch (SQLException e) { + LOGGER.debug("Resetting autocommit failed: {}", e.getMessage()); + } + } + + close(connection); + } + + /** Closes the result set and the statement it belongs to. */ + private static void close(ResultSet resultSet) { + Statement statement = null; + + try { + statement = resultSet.getStatement(); + } catch (SQLException e) { + // the result set is closed below regardless + } + + close((AutoCloseable) resultSet); + close(statement); + } + + private static void close(AutoCloseable closeable) { + if (Objects.nonNull(closeable)) { + try { + closeable.close(); + } catch (Exception e) { + LOGGER.debug("Closing {} failed: {}", closeable.getClass().getSimpleName(), e.getMessage()); + } + } + } + @Override public Connection getConnection() { return leaseConnection(); diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlConnectorRx.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlConnectorRx.java index c8f5fa333..0e33f3fce 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlConnectorRx.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlConnectorRx.java @@ -51,7 +51,6 @@ import java.util.concurrent.Semaphore; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; -import org.davidmoten.rxjava3.jdbc.Database; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.slf4j.MDC; @@ -80,7 +79,6 @@ public class SqlConnectorRx extends AbstractVolatilePolling implements SqlConnec private final AtomicInteger refCounter; private final boolean asyncStartup; - private Database session; private HikariDataSource dataSource; private SqlClient sqlClient; private Throwable connectionError; @@ -166,10 +164,8 @@ public void start() { try { HikariConfig hikariConfig = createHikariConfig(); this.dataSource = new HikariDataSource(hikariConfig); - this.session = createSession(dataSource); this.sqlClient = new SqlClientRx( - session, dataSource, dbmsAdapters.get(connectionInfo.getDialect()), dbmsAdapters.getDialect(connectionInfo.getDialect()), @@ -192,13 +188,6 @@ public void stop() { // ignore } } - if (Objects.nonNull(session)) { - try { - session.close(); - } catch (Throwable e) { - // ignore - } - } if (Objects.nonNull(dataSource)) { try { dataSource.close(); @@ -350,10 +339,6 @@ private HikariConfig createHikariConfig() { return config; } - private Database createSession(HikariDataSource dataSource) { - return Database.fromBlocking(dataSource); - } - private static long getInitFailTimeout(ConnectionInfoSql connectionInfo) { return parseMs(Objects.requireNonNullElse(connectionInfo.getPool().getInitFailTimeout(), "1")); } diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlDbmsAdapterGpkg.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlDbmsAdapterGpkg.java index 735e4bce0..9529f9c68 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlDbmsAdapterGpkg.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlDbmsAdapterGpkg.java @@ -38,7 +38,6 @@ import java.util.Objects; import java.util.Optional; import javax.sql.DataSource; -import org.davidmoten.rxjava3.jdbc.pool.DatabaseType; import org.immutables.value.Value; import org.sqlite.SQLiteConnection; import org.sqlite.SQLiteDataSource; @@ -140,11 +139,6 @@ public Optional getInitSql(ConnectionInfoSql connectionInfo) { return Optional.of("SELECT CASE CheckGeoPackageMetaData() WHEN 1 THEN EnableGpkgMode() END;"); } - @Override - public DatabaseType getRxType() { - return DatabaseType.SQLITE; - } - @Override public List getDefaultSchemas() { return List.of(); diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlDbmsAdapterPgis.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlDbmsAdapterPgis.java index ed5d9929a..dbbdb1f51 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlDbmsAdapterPgis.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/infra/db/SqlDbmsAdapterPgis.java @@ -35,7 +35,6 @@ import java.util.Optional; import java.util.stream.Collectors; import javax.sql.DataSource; -import org.davidmoten.rxjava3.jdbc.pool.DatabaseType; import org.immutables.value.Value; import org.postgresql.ds.PGSimpleDataSource; import org.slf4j.Logger; @@ -144,11 +143,6 @@ public Optional getInitSql(ConnectionInfoSql connectionInfo) { return Optional.of(initSql); } - @Override - public DatabaseType getRxType() { - return DatabaseType.POSTGRES; - } - @Override public List getDefaultSchemas() { return List.of("public"); diff --git a/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/SqlClientRxSpec.groovy b/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/SqlClientRxSpec.groovy index f8db01dd4..3aa87d095 100644 --- a/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/SqlClientRxSpec.groovy +++ b/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/infra/db/SqlClientRxSpec.groovy @@ -7,30 +7,40 @@ */ package de.ii.xtraplatform.features.sql.infra.db +import de.ii.xtraplatform.features.sql.domain.ImmutableSqlQueryOptions import de.ii.xtraplatform.features.sql.domain.SqlDbmsAdapter import de.ii.xtraplatform.features.sql.domain.SqlDialect import de.ii.xtraplatform.features.sql.domain.SqlQueryOptions -import org.davidmoten.rxjava3.jdbc.ConnectionProvider -import org.davidmoten.rxjava3.jdbc.Database +import de.ii.xtraplatform.features.sql.domain.SqlRow +import de.ii.xtraplatform.streams.app.SourceDefault +import io.reactivex.rxjava3.core.Flowable +import io.reactivex.rxjava3.subscribers.TestSubscriber import spock.lang.Specification import javax.sql.DataSource import java.sql.Connection +import java.sql.ResultSet import java.sql.SQLException import java.sql.Statement import java.util.concurrent.CompletionException /** - * Locks the connection lifecycle of everything in {@link SqlClientRx} that is not a streamed read: - * a connection is leased from the pool directly and returned to it in every case, so no code path - * outside the reads depends on the rxjava3-jdbc reference counting. + * Locks the connection lifecycle of {@link SqlClientRx}: every connection is leased from the pool + * when it is needed and returned in every case - rows exhausted, statement failed, consumer + * cancelled - so the pool cannot be drained by a read or a statement that ends abnormally. */ class SqlClientRxSpec extends Specification { + static final SqlQueryOptions PAGED = SqlQueryOptions.withColumnTypes(String.class) + static final SqlQueryOptions STREAMED = + new ImmutableSqlQueryOptions.Builder().from(PAGED).fetchSize(2).build() + DataSource dataSource Connection connection Statement statement + ResultSet resultSet boolean leaseFails + boolean executeFails SqlClientRx sqlClient @@ -38,22 +48,123 @@ class SqlClientRxSpec extends Specification { dataSource = Mock(DataSource) connection = Mock(Connection) statement = Mock(Statement) + resultSet = Mock(ResultSet) leaseFails = false + executeFails = false dataSource.getConnection() >> { if (leaseFails) throw new SQLException('Connection is not available, request timed out after 30000ms') return connection } connection.createStatement() >> statement + statement.executeQuery(_ as String) >> { + if (executeFails) throw new SQLException('relation does not exist', '42P01') + return resultSet + } + resultSet.getStatement() >> statement + + sqlClient = new SqlClientRx(dataSource, Mock(SqlDbmsAdapter), Mock(SqlDialect), Optional.empty()) + } + + def 'a streamed read runs in a transaction, closes statement and result set and returns the connection'() { + given: + rows('a', 'b') + + when: + TestSubscriber subscriber = read('SELECT id FROM t', STREAMED) + + then: + subscriber.assertValueCount(2) + subscriber.assertComplete() + 1 * connection.setAutoCommit(false) + 1 * statement.setFetchSize(2) + 1 * resultSet.close() + 1 * statement.close() + 1 * connection.rollback() + 1 * connection.setAutoCommit(true) + 1 * connection.close() + } + + def 'a paged read runs with autocommit and returns the connection'() { + given: + rows('a') + + when: + TestSubscriber subscriber = read('SELECT id FROM t', PAGED) + + then: + subscriber.assertValueCount(1) + subscriber.assertComplete() + 0 * connection.setAutoCommit(_) + 0 * statement.setFetchSize(_) + 0 * connection.rollback() + 1 * connection.close() + } + + def 'a read cancelled by the consumer returns the connection'() { + given: + rows('a', 'b', 'c') + + when: + TestSubscriber subscriber = Flowable.fromPublisher(publisher('SELECT id FROM t', STREAMED)).take(1).test() + + then: + subscriber.assertValueCount(1) + subscriber.assertComplete() + 1 * resultSet.close() + 1 * connection.rollback() + 1 * connection.close() + } + + def 'an error while reading rows is delivered and the connection is returned'() { + given: + resultSet.next() >> { throw new SQLException('terminating connection due to administrator command', '57P01') } + + when: + TestSubscriber subscriber = read('SELECT id FROM t', STREAMED) - Database database = Database.fromBlocking(new ConnectionProvider() { - @Override - Connection get() { return connection } + then: + subscriber.assertError(SQLException) + 1 * resultSet.close() + 1 * connection.close() + } - @Override - void close() {} - }) + def 'a statement that fails to execute is delivered as an error and the connection is returned'() { + given: + executeFails = true - sqlClient = new SqlClientRx(database, dataSource, Mock(SqlDbmsAdapter), Mock(SqlDialect), Optional.empty()) + when: + TestSubscriber subscriber = read('SELECT id FROM missing', STREAMED) + + then: + subscriber.assertError(SQLException) + 1 * statement.close() + 1 * connection.close() + } + + def 'a connection that cannot be leased for a read fails the read'() { + given: + leaseFails = true + + when: + TestSubscriber subscriber = read('SELECT id FROM t', PAGED) + + then: + subscriber.assertError(SQLException) + 0 * connection.close() + } + + def 'run reads all rows and returns the connection'() { + given: + rows('a', 'b', 'c') + + when: + def result = sqlClient.run('SELECT id FROM t', PAGED).join() + + then: + result.size() == 3 + 1 * resultSet.close() + 1 * statement.close() + 1 * connection.close() } def 'a statement without a result runs on a pooled connection that is closed afterwards'() { @@ -103,4 +214,18 @@ class SqlClientRxSpec extends Specification { e.cause instanceof SQLException e.message.contains('not available') } + + private void rows(String... ids) { + List next = ids.collect { true } + [false] + resultSet.next() >>> next + resultSet.getString(1) >>> (ids as List) + } + + private TestSubscriber read(String sql, SqlQueryOptions options) { + return Flowable.fromPublisher(publisher(sql, options)).test() + } + + private org.reactivestreams.Publisher publisher(String sql, SqlQueryOptions options) { + return ((SourceDefault) sqlClient.getSourceStream(sql, options)).getPublisher() + } }