From e8cea8735ae889c6d3d685a917230d7019394a10 Mon Sep 17 00:00:00 2001 From: Clemens Portele Date: Sun, 28 Jun 2026 12:09:26 +0200 Subject: [PATCH 1/3] features: single-shot and paged execution modes for multi-queries A multi-feature query can run single-shot - one pass per (sub-query, table), with no meta query, no chunking and no key-range window - using a server-side cursor to keep memory bounded; an optional per-sub-query maximum bounds each sub-query. When paged, numberMatched is computed across all sub-queries (the invariant total) while value queries are executed only for the sub-queries needed for the page. --- .../features/sql/app/FeatureDecoderSql.java | 12 +- .../sql/app/FeatureQueryEncoderSql.java | 70 +++++++---- .../features/sql/domain/SqlConnector.java | 36 ++++-- .../features/sql/domain/SqlQueryBatch.java | 21 ++++ .../features/sql/domain/SqlQueryOptions.java | 10 ++ .../features/sql/infra/db/SqlClientRx.java | 36 ++++-- .../sql/app/ResultSetPagingSpec.groovy | 111 ++++++++++++++++++ .../features/domain/MultiFeatureQuery.java | 17 ++- 8 files changed, 265 insertions(+), 48 deletions(-) create mode 100644 xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/app/ResultSetPagingSpec.groovy diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/FeatureDecoderSql.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/FeatureDecoderSql.java index f246e7a7d..517bbaccd 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/FeatureDecoderSql.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/FeatureDecoderSql.java @@ -145,10 +145,14 @@ public void onPush(SqlRow sqlRow) { private void handleMetaRow(SqlRowMeta sqlRow) { - context - .metadata() - .numberReturned( - context.metadata().getNumberReturned().orElse(0) + sqlRow.getNumberReturned()); + // a negative numberReturned marks it as not computed (single-shot/unpaged); leave it unset so + // the encoder omits numberReturned instead of reporting 0 + if (sqlRow.getNumberReturned() >= 0) { + context + .metadata() + .numberReturned( + context.metadata().getNumberReturned().orElse(0) + sqlRow.getNumberReturned()); + } if (sqlRow.getNumberMatched().isPresent()) { context .metadata() diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/FeatureQueryEncoderSql.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/FeatureQueryEncoderSql.java index 086cac6a3..9fd955923 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/FeatureQueryEncoderSql.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/FeatureQueryEncoderSql.java @@ -55,6 +55,7 @@ public class FeatureQueryEncoderSql implements FeatureQueryEncoder> allQueryTemplates, @@ -65,6 +66,7 @@ public FeatureQueryEncoderSql( this.allQueryTemplatesMutations = allQueryTemplatesMutations; this.chunkSize = queryGeneratorSettings.getChunkSize(); this.geometryAsWkb = queryGeneratorSettings.getGeometryAsWkb(); + this.computeNumberMatched = queryGeneratorSettings.getComputeNumberMatched(); this.sqlDialect = sqlDialect; } @@ -114,6 +116,7 @@ private SqlQueryBatch encode(FeatureQuery query, Map additionalQ query, additionalQueryParameters, query.returnsSingleFeature(), + false, 0))) .flatMap(s -> s) .collect(Collectors.toList()); @@ -131,7 +134,13 @@ private SqlQueryBatch encode(FeatureQuery query, Map additionalQ private SqlQueryBatch encode( MultiFeatureQuery query, Map additionalQueryParameters) { - int chunks = (query.getLimit() / chunkSize) + (query.getLimit() % chunkSize > 0 ? 1 : 0); + // A multi-query that does not support paging is executed single-shot: every matching row is + // read in one pass per (sub-query, table), with no meta query, chunking or key-range window. + // This drops numberReturned/numberMatched in exchange for far fewer statements and round-trips + // at large sizes (an optional per-sub-query maximum may still cap each sub-query). + boolean unpaged = !query.getSupportPaging() && !query.hitsOnly(); + int chunks = + unpaged ? 1 : (query.getLimit() / chunkSize) + (query.getLimit() % chunkSize > 0 ? 1 : 0); List querySets = IntStream.range(0, query.getQueries().size()) @@ -156,7 +165,8 @@ private SqlQueryBatch encode( typeQuery, query, additionalQueryParameters, - false, + unpaged, + unpaged, queryIndex))) .flatMap(s -> s); }) @@ -168,6 +178,9 @@ private SqlQueryBatch encode( .limit(query.getLimit()) .offset(query.getOffset()) .chunkSize(chunkSize) + .isUnpaged(unpaged) + // when paging, count every sub-query so numberMatched is the full invariant total + .isComputeNumberMatched(query.getSupportPaging() && computeNumberMatched) .build() .withQuerySets(querySets); } @@ -181,14 +194,14 @@ private SqlQuerySet createQuerySet( Query query, Map additionalQueryParameters, boolean skipMetaQuery, + boolean unpaged, int queryIndex) { List sortKeys = transformSortKeys(typeQuery.getSortKeys(), queryTemplates.getMapping()); boolean useMinMaxKeys = queryTemplates.getMapping().getMainTable().isSortKeyUnique(); - // a multi-query may opt out of computing numberMatched to avoid a count query per sub-query - boolean computeNumberMatched = - !(query instanceof MultiFeatureQuery) - || ((MultiFeatureQuery) query).getComputeNumberMatched(); + // a paged multi-query computes numberMatched; a single-shot one (no paging) does not + boolean supportPaging = + !(query instanceof MultiFeatureQuery) || ((MultiFeatureQuery) query).getSupportPaging(); BiFunction> metaQuery = (maxLimit, skipped) -> @@ -209,27 +222,42 @@ private SqlQuerySet createQuerySet( query.hitsOnly(), // numberMatched is invariant across chunks, so compute it only on the // first chunk of each collection; later chunks reuse that value - chunk == 0 && computeNumberMatched)); + chunk == 0 && supportPaging)); TriFunction> valueQueries = (metaResult, maxLimit, skipped) -> queryTemplates.getValueQueryTemplates().stream() .map( valueQueryTemplate -> - valueQueryTemplate.generateValueQuery( - Math.min(limit, maxLimit), - Math.max(0L, offset - skipped), - sortKeys, - typeQuery.getFilter(), - typeQuery.forceSimpleFeatureGeometry(), - (useMinMaxKeys - && ((Objects.nonNull(metaResult.getMinKey()) - && Objects.nonNull(metaResult.getMaxKey())) - || metaResult.getNumberReturned() == 0)) - ? Optional.of( - Tuple.of(metaResult.getMinKey(), metaResult.getMaxKey())) - : Optional.empty(), - additionalQueryParameters)); + // single-shot reads matching rows in one pass: no offset and no key-range + // window (which would otherwise constrain the result set to the meta + // query's minKey/maxKey); an optional per-sub-query maximum caps each + // sub-query (0 = no limit) + unpaged + ? valueQueryTemplate.generateValueQuery( + query instanceof MultiFeatureQuery + ? ((MultiFeatureQuery) query).getMaxFeaturesPerSubQuery() + : 0L, + 0L, + sortKeys, + typeQuery.getFilter(), + typeQuery.forceSimpleFeatureGeometry(), + Optional.empty(), + additionalQueryParameters) + : valueQueryTemplate.generateValueQuery( + Math.min(limit, maxLimit), + Math.max(0L, offset - skipped), + sortKeys, + typeQuery.getFilter(), + typeQuery.forceSimpleFeatureGeometry(), + (useMinMaxKeys + && ((Objects.nonNull(metaResult.getMinKey()) + && Objects.nonNull(metaResult.getMaxKey())) + || metaResult.getNumberReturned() == 0)) + ? Optional.of( + Tuple.of(metaResult.getMinKey(), metaResult.getMaxKey())) + : Optional.empty(), + additionalQueryParameters)); // reuse SchemaSql instances instead of copying them; this is expensive and unnecessary, since // they are immutable diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlConnector.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlConnector.java index 778c4d04d..3b0d8f5ff 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlConnector.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlConnector.java @@ -57,11 +57,13 @@ class Paging { private long lastNumberReturned; private long lastNumberSkipped; private boolean noOffset; + private final boolean computeNumberMatched; - public Paging(long limit, long offset, long chunkSize) { + public Paging(long limit, long offset, long chunkSize, boolean computeNumberMatched) { this.limit = limit; this.offset = offset; this.chunkSize = chunkSize; + this.computeNumberMatched = computeNumberMatched; this.featureCountdown = limit; this.numberSkipped = 0L; @@ -75,9 +77,14 @@ Optional> get(String currentTable) { long found = lastNumberReturned + lastNumberSkipped; // Once the limit is reached or the current collection is exhausted (its last chunk returned - // fewer rows than the chunk size), no further meta query is needed: there are no more rows to - // read and numberMatched was already computed on the collection's first chunk. + // fewer rows than the chunk size), no further rows need to be read. When numberMatched is + // computed, a not-yet-counted sub-query is still given a count-only meta query (maxLimit 0) + // so the reported total covers every sub-query, not only those contributing to the page; its + // value query is skipped downstream because the meta reports numberReturned 0. if (featureCountdown <= 0 || (Objects.equals(lastTable, currentTable) && found < chunkSize)) { + if (computeNumberMatched && !Objects.equals(lastTable, currentTable)) { + return Optional.of(Tuple.of(0L, 0L)); + } return Optional.empty(); } @@ -109,7 +116,11 @@ void register(String currentTable, SqlRowMeta metaResult) { default Reactive.Source getSourceStream( SqlQueryBatch queryBatch, SqlQueryOptions options) { Paging paging = - new Paging(queryBatch.getLimit(), queryBatch.getOffset(), queryBatch.getChunkSize()); + new Paging( + queryBatch.getLimit(), + queryBatch.getOffset(), + queryBatch.getChunkSize(), + queryBatch.isComputeNumberMatched()); Source sqlRowSource1 = Source.iterable(queryBatch.getQuerySets()) @@ -196,10 +207,17 @@ default Reactive.Source getSourceStream( .via( Transformer.flatMap( plan -> { - if (queryBatch.isSingleFeature()) { + if (queryBatch.isSingleFeature() || queryBatch.isUnpaged()) { + boolean unpaged = queryBatch.isUnpaged(); List querySets = queryBatch.getQuerySets(); + // a single-shot query computes neither count; -1 marks both as absent (the + // decoder leaves numberReturned unset and numberMatched stays empty) + // instead of reporting 0 ImmutableSqlRowMeta sqlRowMeta = - getMetaQueryResult(0L, 0L, 0L, 0L, -1L).build(); + getMetaQueryResult(0L, 0L, unpaged ? -1L : 0L, unpaged ? -1L : 0L, -1L) + .build(); + // a server-side cursor keeps memory bounded while streaming all rows + int fetchSize = unpaged ? (int) queryBatch.getChunkSize() : 0; return Source.iterable( IntStream.range(0, querySets.size()) .boxed() @@ -235,6 +253,7 @@ default Reactive.Source getSourceStream( querySets .get(index) .getQueryIndex()) + .fetchSize(fetchSize) .build())) .toArray( (IntFunction[]>) Source[]::new); @@ -247,11 +266,14 @@ default Reactive.Source getSourceStream( List querySets = plan.first(); SqlRowMeta aggregatedMetaResult = plan.second().get(0); List metaResults = plan.second().subList(1, plan.second().size()); + // the value phase only reads window sub-queries; count-only ones are skipped + // via the numberReturned<=0 guard below, so it needs no count-only handling Paging paging2 = new Paging( queryBatch.getLimit(), queryBatch.getOffset(), - queryBatch.getChunkSize()); + queryBatch.getChunkSize(), + false); int[] i = {0}; if (options.isHitsOnly()) { diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlQueryBatch.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlQueryBatch.java index 9763e666e..15acb2d6d 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlQueryBatch.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlQueryBatch.java @@ -24,5 +24,26 @@ default boolean isSingleFeature() { return false; } + /** + * Single-shot (unpaged) mode: every matching row is read in one pass per (sub-query, table) + * without a meta query, without chunking, and without a key-range window. {@code limit} does not + * page and {@code numberReturned}/{@code numberMatched} are not computed. + */ + @Value.Default + default boolean isUnpaged() { + return false; + } + + /** + * Whether {@code numberMatched} is reported. When enabled for a multi-query, the count is + * computed for every sub-query (not only those contributing to the current page) so that the + * reported {@code numberMatched} is the invariant total across all sub-queries; the value queries + * are still executed only for the sub-queries needed for the page. + */ + @Value.Default + default boolean isComputeNumberMatched() { + return false; + } + List getQuerySets(); } diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlQueryOptions.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlQueryOptions.java index 30e949302..d2ba5ebb1 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlQueryOptions.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlQueryOptions.java @@ -79,6 +79,16 @@ default int getChunkSize() { return 1000; } + /** + * JDBC fetch size for the result set. When greater than 0, the query is executed inside a + * transaction so the database driver can use a server-side cursor and stream rows instead of + * buffering the whole result set in memory. Used for single-shot (unpaged) queries. + */ + @Value.Default + default int getFetchSize() { + return 0; + } + @Value.Default default boolean isGeometryWkb() { return false; 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 39693d796..dd671d55f 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 @@ -98,21 +98,31 @@ public Reactive.Source getSourceStream(String query, SqlQueryOptions opt } List logBuffer = new ArrayList<>(5); - // TODO encapsulating the query in a transaction is a workaround for what appears to be a bug in - // rxjava3-jdbc, see https://github.com/interactive-instruments/ldproxy/issues/1293 - Flowable flowable = - session - .select(query) - .get( - resultSet -> { - SqlRow row = new SqlRowVals(collator).read(resultSet, options); + 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); - } + if (LOGGER.isDebugEnabled(MARKER.SQL_RESULT) && logBuffer.size() < 10) { + logBuffer.add(row); + } - return 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 + Flowable flowable = + options.getFetchSize() > 0 + ? session + .select(query) + .transacted() + .fetchSize(options.getFetchSize()) + .valuesOnly() + .get(mapper) + : session.select(query).get(mapper); // TODO: prettify, see // https://github.com/slick/slick/blob/main/slick/src/main/scala/slick/jdbc/StatementInvoker.scala diff --git a/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/app/ResultSetPagingSpec.groovy b/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/app/ResultSetPagingSpec.groovy new file mode 100644 index 000000000..10057dc87 --- /dev/null +++ b/xtraplatform-features-sql/src/test/groovy/de/ii/xtraplatform/features/sql/app/ResultSetPagingSpec.groovy @@ -0,0 +1,111 @@ +/* + * Copyright 2026 interactive instruments GmbH + * + * This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at http://mozilla.org/MPL/2.0/. + */ +package de.ii.xtraplatform.features.sql.app + +import com.google.common.collect.ImmutableMap +import de.ii.xtraplatform.cql.app.CqlImpl +import de.ii.xtraplatform.cql.domain.Eq +import de.ii.xtraplatform.cql.domain.ImmutableInResultSet +import de.ii.xtraplatform.cql.domain.InResultSet +import de.ii.xtraplatform.cql.domain.Property +import de.ii.xtraplatform.cql.domain.ScalarLiteral +import de.ii.xtraplatform.crs.domain.OgcCrs +import de.ii.xtraplatform.features.domain.FeatureSchemaFixtures +import de.ii.xtraplatform.features.domain.MappingOperationResolver +import de.ii.xtraplatform.features.domain.MappingRuleFixtures +import de.ii.xtraplatform.features.domain.SortKey +import de.ii.xtraplatform.features.domain.Tuple +import de.ii.xtraplatform.features.json.app.DecoderFactoryJson +import de.ii.xtraplatform.features.sql.domain.ImmutableQueryGeneratorSettings +import de.ii.xtraplatform.features.sql.domain.ImmutableSqlPathDefaults +import de.ii.xtraplatform.features.sql.domain.SqlDialectPgis +import de.ii.xtraplatform.features.sql.domain.SqlPathParser +import de.ii.xtraplatform.features.sql.domain.SqlQueryMapping +import spock.lang.Shared +import spock.lang.Specification + +import java.util.function.Function + +/** + * A paged query expression (supportPaging=true) executes the chunked/keyset path while its + * sub-queries may carry a result-set ('inResultSet') filter. This verifies that the value query + * composes both: the result-set membership sub-query is kept and the keyset paging window is + * applied alongside it. A single-shot query (supportPaging=false) keeps the membership but adds no + * paging. + */ +class ResultSetPagingSpec extends Specification { + + @Shared + Map mappings = [:] + @Shared + SqlQueryTemplatesDeriver deriver + @Shared + SqlMappingDeriver mappingDeriver + @Shared + MappingOperationResolver mappingOperationResolver + + def setupSpec() { + def defaults = new ImmutableSqlPathDefaults.Builder().build() + def cql = new CqlImpl() + def pathParser = new SqlPathParser(defaults, cql, Map.of("JSON", new DecoderFactoryJson(), "EXPRESSION", new DecoderFactorySqlExpression())) + mappingDeriver = new SqlMappingDeriver(pathParser, new ImmutableQueryGeneratorSettings.Builder().build()) + mappingOperationResolver = new MappingOperationResolver() + + ["simple", "value_array"].each { name -> + def schema = FeatureSchemaFixtures.fromYaml(name) + def resolvedSchema = schema.accept(mappingOperationResolver, List.of()) + def rules = MappingRuleFixtures.fromYaml(name) + mappings[name] = mappingDeriver.derive(rules, resolvedSchema).get(0) + } + + // a filter encoder that can resolve result-set producers, like the one used at runtime + def filterEncoder = new FilterEncoderSql(OgcCrs.CRS84, new SqlDialectPgis(), null, null, cql, List.of(), null, + { type -> Optional.ofNullable(mappings[type]) } as Function) + deriver = new SqlQueryTemplatesDeriver(filterEncoder, new SqlDialectPgis(), true, false, Optional.empty()) + } + + static InResultSet inResultSet(String producerType, de.ii.xtraplatform.cql.domain.Cql2Expression producerFilter) { + return new ImmutableInResultSet.Builder() + .from(InResultSet.of("id", "s1")) + .producerType(producerType) + .producerFilter(producerFilter) + .build() + } + + List mainValueQuery(int limit, int offset, Optional> minMaxKeys) { + def schema = FeatureSchemaFixtures.fromYaml("value_array") + def resolvedSchema = schema.accept(mappingOperationResolver, List.of()) + def rules = MappingRuleFixtures.fromYaml("value_array") + def templates = mappingDeriver.derive(rules, resolvedSchema).stream().map(deriver.&derive).toList() + def filter = inResultSet("simple", Eq.of(Property.of("id"), ScalarLiteral.of("foo"))) + return templates.get(0).getValueQueryTemplates() + .collect { it.generateValueQuery(limit, offset, [] as List, Optional.of(filter), false, minMaxKeys, ImmutableMap.of()) } + } + + def 'a paged value query keeps the result-set membership and applies the keyset window'() { + + when: 'the main value query is generated with the filter and a keyset window (limit 10, offset 10)' + String mainQuery = mainValueQuery(10, 10, Optional.of(Tuple.of(10, 19))).get(0) + + then: 'the result-set membership sub-query and the keyset window both appear' + mainQuery.contains("_rs_0_s1 AS MATERIALIZED") + mainQuery.contains(">= 10") + mainQuery.contains("<= 19") + } + + def 'a single-shot value query keeps the result-set membership but has no paging window'() { + + when: 'the main value query is generated single-shot (no limit, no keyset)' + String mainQuery = mainValueQuery(0, 0, Optional.> empty()).get(0) + + then: 'the result-set membership is present and there is no LIMIT/OFFSET paging clause' + mainQuery.contains("_rs_0_s1 AS MATERIALIZED") + !mainQuery.contains("LIMIT") + !mainQuery.contains("OFFSET") + } +} diff --git a/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/MultiFeatureQuery.java b/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/MultiFeatureQuery.java index 7700c0de7..30c002383 100644 --- a/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/MultiFeatureQuery.java +++ b/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/MultiFeatureQuery.java @@ -30,11 +30,22 @@ default boolean getDeduplicate() { } /** - * If disabled, {@code numberMatched} is not computed. For a multi-query this avoids a count query - * per sub-query, each of which carries the full (possibly deeply nested) filter. + * If enabled (the default), the query is executed with paging support: {@code numberReturned} and + * {@code numberMatched} are computed and {@code limit}/{@code offset} select a page of the result + * set. If disabled, the query is executed single-shot: all matching features are returned in one + * pass, without meta queries and without {@code numberReturned}/{@code numberMatched}. */ @Value.Default - default boolean getComputeNumberMatched() { + default boolean getSupportPaging() { return true; } + + /** + * In single-shot mode (paging disabled), an optional maximum number of features read per + * sub-query. {@code 0} means no limit. + */ + @Value.Default + default int getMaxFeaturesPerSubQuery() { + return 0; + } } From 7ea5445ed33b8cc0ab9b68c0322d93948fa84f0e Mon Sep 17 00:00:00 2001 From: Clemens Portele Date: Sun, 28 Jun 2026 15:29:16 +0200 Subject: [PATCH 2/3] features: run single-shot multi-query sub-queries concurrently Single-shot multi-queries read every sub-query in one pass with no cross-sub-query dependency, so the sub-queries can now execute concurrently while their rows are still emitted in sub-query order: - the value phase uses an order-preserving concurrent flat-map over the sub-queries (Transformer.flatMapConcurrent); - each sub-query's value queries are subscribed on a worker thread (SqlQueryOptions.isParallel) so the blocking JDBC reads actually overlap instead of running on a single thread; - a shared connection-budget semaphore (sized to the pool) bounds the connections held by the in-flight sub-queries and acquires each sub-query's permits as a block, so concurrent requests cannot deadlock by each holding part of the pool. Paged and single-feature reads keep the serial path. Requires the matching xtraplatform-core operators flatMapConcurrent and guarded. --- gradle/layers.versions.toml | 2 +- .../features/sql/domain/SqlConnector.java | 132 +++++++++++++----- .../features/sql/domain/SqlQueryOptions.java | 10 ++ .../features/sql/infra/db/SqlClientRx.java | 10 ++ .../features/sql/infra/db/SqlConnectorRx.java | 8 ++ 5 files changed, 123 insertions(+), 39 deletions(-) diff --git a/gradle/layers.versions.toml b/gradle/layers.versions.toml index c4d40b332..2c4e079ee 100644 --- a/gradle/layers.versions.toml +++ b/gradle/layers.versions.toml @@ -1,4 +1,4 @@ [versions] -xtraplatform-core = '7.0.0-SNAPSHOT' +xtraplatform-core = '7.0.0-reactive-concurrent-flatmap-SNAPSHOT' xtraplatform-native = '2.6.0-SNAPSHOT' diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlConnector.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlConnector.java index 3b0d8f5ff..cf807662b 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlConnector.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlConnector.java @@ -22,6 +22,7 @@ import java.util.Objects; import java.util.Optional; import java.util.OptionalLong; +import java.util.concurrent.Semaphore; import java.util.function.Function; import java.util.function.IntFunction; import java.util.stream.Collectors; @@ -34,6 +35,12 @@ public interface SqlConnector int getMaxConnections(); + /** + * Shared budget of connection permits (sized to the connection pool) that bounds how many + * connections the concurrently running single-shot sub-queries hold at once, across all requests. + */ + Semaphore getConnectionBudget(); + int getMinConnections(); int getQueueSize(); @@ -111,6 +118,9 @@ void register(String currentTable, SqlRowMeta metaResult) { } } + // Per-sub-query row buffer for the concurrent single-shot value phase (see getSourceStream). + int UNPAGED_SUBQUERY_PREFETCH = 256; + // TODO: simplify, class SqlQueryRunner, remove options, singleFeature @Override default Reactive.Source getSourceStream( @@ -218,48 +228,94 @@ default Reactive.Source getSourceStream( .build(); // a server-side cursor keeps memory bounded while streaming all rows int fetchSize = unpaged ? (int) queryBatch.getChunkSize() : 0; + Function> valuePhase = + index -> { + int[] i = {0}; + Source[] sqlRows = + querySets + .get(index) + .getValueQueries() + .apply(sqlRowMeta, 0L, 0L) + .map( + valueQuery -> + getSqlClient() + .getSourceStream( + valueQuery, + new ImmutableSqlQueryOptions.Builder() + .from(options) + .tableSchema( + querySets + .get(index) + .getTableSchemas() + .get(i[0])) + .type( + querySets + .get(index) + .getOptions() + .getType()) + .containerPriority(i[0]++) + .queryIndex( + querySets.get(index).getQueryIndex()) + .fetchSize(fetchSize) + .isParallel(unpaged) + .build())) + .toArray((IntFunction[]>) Source[]::new); + + Source merged = mergeAndSort(sqlRows); + + if (!unpaged) { + return merged; + } + + // Bound the connections held by concurrently running sub-queries: a + // sub-query needs all its table connections at once, so acquire them + // as + // a block before its rows are read and release them when it + // terminates. + // The semaphore is shared across requests, so concurrent searches can + // never deadlock by each grabbing part of the pool. + int tables = querySets.get(index).getTableSchemas().size(); + return Reactive.Source.guarded( + () -> getConnectionBudget().acquireUninterruptibly(tables), + () -> getConnectionBudget().release(tables), + merged); + }; + + // Single-shot multi-queries read every sub-query in one pass with no + // cross-sub-query dependency (the result sets are already materialized), so + // the sub-queries can run concurrently. Their rows are still emitted + // strictly + // in sub-query order, which the decoder requires; only the JDBC reads + // overlap. + // Concurrency is bounded so the sub-queries in flight never request more + // than + // the connection pool can serve. Single-feature reads keep the serial path. + Transformer valueTransformer; + if (unpaged) { + int maxTables = + Math.max( + 1, + querySets.stream() + .mapToInt(qs -> qs.getTableSchemas().size()) + .max() + .orElse(1)); + int maxConcurrency = + Math.max( + 1, + Math.min( + querySets.size(), (getMaxConnections() - 1) / maxTables)); + valueTransformer = + Transformer.flatMapConcurrent( + valuePhase, maxConcurrency, UNPAGED_SUBQUERY_PREFETCH); + } else { + valueTransformer = Transformer.flatMap(valuePhase); + } + return Source.iterable( IntStream.range(0, querySets.size()) .boxed() .collect(Collectors.toList())) - .via( - Transformer.flatMap( - index -> { - int[] i = {0}; - Source[] sqlRows = - querySets - .get(index) - .getValueQueries() - .apply(sqlRowMeta, 0L, 0L) - .map( - valueQuery -> - getSqlClient() - .getSourceStream( - valueQuery, - new ImmutableSqlQueryOptions.Builder() - .from(options) - .tableSchema( - querySets - .get(index) - .getTableSchemas() - .get(i[0])) - .type( - querySets - .get(index) - .getOptions() - .getType()) - .containerPriority(i[0]++) - .queryIndex( - querySets - .get(index) - .getQueryIndex()) - .fetchSize(fetchSize) - .build())) - .toArray( - (IntFunction[]>) Source[]::new); - - return mergeAndSort(sqlRows); - })) + .via(valueTransformer) .prepend(Source.single(sqlRowMeta)); } diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlQueryOptions.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlQueryOptions.java index d2ba5ebb1..8ef976657 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlQueryOptions.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlQueryOptions.java @@ -89,6 +89,16 @@ default int getFetchSize() { return 0; } + /** + * When enabled, the query is subscribed on a worker thread so that several such queries can + * execute concurrently (the blocking JDBC connection provider otherwise runs every query on the + * single subscribing thread). Used for the concurrent single-shot value phase. + */ + @Value.Default + default boolean isParallel() { + return false; + } + @Value.Default default boolean isGeometryWkb() { return false; 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 dd671d55f..1afe2c7f9 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 @@ -21,6 +21,7 @@ import de.ii.xtraplatform.streams.domain.Reactive; import de.ii.xtraplatform.streams.domain.Reactive.Transformer; import io.reactivex.rxjava3.core.Flowable; +import io.reactivex.rxjava3.schedulers.Schedulers; import java.sql.Connection; import java.sql.ResultSet; import java.sql.SQLException; @@ -159,6 +160,15 @@ 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. + if (options.isParallel()) { + flowable = flowable.subscribeOn(Schedulers.io()); + } + return Reactive.Source.publisher(flowable); } 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 5f593ced5..3a81c4864 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 @@ -48,6 +48,7 @@ import java.util.Map.Entry; import java.util.Objects; import java.util.Optional; +import java.util.concurrent.Semaphore; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import org.davidmoten.rxjava3.jdbc.Database; @@ -72,6 +73,7 @@ public class SqlConnectorRx extends AbstractVolatilePolling implements SqlConnec private final int maxConnections; private final int minConnections; private final int queueSize; + private final Semaphore connectionBudget; private final Path dataDir; private final String applicationName; private final String providerId; @@ -105,6 +107,7 @@ public SqlConnectorRx( connectionInfo.getPool().getMinConnections() >= 0 ? connectionInfo.getPool().getMinConnections() : maxConnections; + this.connectionBudget = new Semaphore(Math.max(1, maxConnections), true); // int capacity = maxConnections / maxQueries; // TODO @@ -138,6 +141,11 @@ public int getMaxConnections() { return maxConnections; } + @Override + public Semaphore getConnectionBudget() { + return connectionBudget; + } + @Override public int getMinConnections() { return minConnections; From 136a4f3694b0671a79e6b4afbd8310b50de750f1 Mon Sep 17 00:00:00 2001 From: Andreas Zahnen Date: Fri, 10 Jul 2026 12:54:29 +0200 Subject: [PATCH 3/3] upgrade xtraplatform --- gradle/layers.versions.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/gradle/layers.versions.toml b/gradle/layers.versions.toml index 2c4e079ee..c4d40b332 100644 --- a/gradle/layers.versions.toml +++ b/gradle/layers.versions.toml @@ -1,4 +1,4 @@ [versions] -xtraplatform-core = '7.0.0-reactive-concurrent-flatmap-SNAPSHOT' +xtraplatform-core = '7.0.0-SNAPSHOT' xtraplatform-native = '2.6.0-SNAPSHOT'