This is an automated email from the ASF dual-hosted git repository.
sergehuber pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/unomi.git
The following commit(s) were added to refs/heads/master by this push:
new 03deaf9de UNOMI-1020: Count aggregate matches exactly on the optimized
path (#928)
03deaf9de is described below
commit 03deaf9de44887ee5c936d537c558c1015449ecf
Author: Serge Huber <[email protected]>
AuthorDate: Tue Oct 27 20:57:40 2026 +0100
UNOMI-1020: Count aggregate matches exactly on the optimized path (#928)
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../apache/unomi/itests/PersistenceServiceIT.java | 37 ++++++++++++++++++++++
.../ElasticSearchPersistenceServiceImpl.java | 21 ++++--------
.../OpenSearchPersistenceServiceImpl.java | 22 ++++---------
.../unomi/persistence/spi/PersistenceService.java | 3 +-
.../unomi/rest/endpoints/QueryServiceEndPoint.java | 2 +-
5 files changed, 54 insertions(+), 31 deletions(-)
diff --git
a/itests/src/test/java/org/apache/unomi/itests/PersistenceServiceIT.java
b/itests/src/test/java/org/apache/unomi/itests/PersistenceServiceIT.java
index 75ac30fe7..b7407678b 100644
--- a/itests/src/test/java/org/apache/unomi/itests/PersistenceServiceIT.java
+++ b/itests/src/test/java/org/apache/unomi/itests/PersistenceServiceIT.java
@@ -437,6 +437,43 @@ public class PersistenceServiceIT extends BaseIT {
Assert.assertEquals(Long.valueOf(2L), byLastName.get("_filtered"));
}
+ /**
+ * Above 10,000 matches, the optimized aggregate must still report the
exact number of matching items as
+ * _filtered, even when the bucket limit leaves some values out.
+ */
+ @Test
+ public void testOptimizedAggregateCountsMoreThan10000MatchesExactly()
throws InterruptedException {
+ String marker = "agg-many-" + UUID.randomUUID();
+ int count = 10_005;
+ Condition matches = firstNameEquals(marker);
+ try {
+ for (int i = 0; i < count; i++) {
+ Profile profile = new Profile(marker + "-" + i);
+ profile.setProperty("firstName", marker);
+ profile.setProperty("lastName", "name-" + (i % 3));
+ persistenceService.save(profile, true);
+ }
+ keepTrying("All " + count + " profiles should be indexed", () -> {
+ persistenceService.refreshIndex(Profile.class);
+ return persistenceService.queryCount(matches,
Profile.ITEM_TYPE);
+ }, indexed -> indexed == count, DEFAULT_TRYING_TIMEOUT,
120);
+
+ // Three last names but a bucket limit of 1, so the buckets only
hold about a third of the matches
+ Map<String, Long> byLastName =
persistenceService.aggregateWithOptimizedQuery(matches,
+ new TermsAggregate("properties.lastName"),
Profile.ITEM_TYPE, 1);
+ Assert.assertEquals(Long.valueOf(count),
byLastName.get("_filtered"));
+
+ // The non-optimized path counts the tenant (_all) and the matches
(_filtered) exactly too
+ Map<String, Long> nonOptimized =
persistenceService.aggregateQuery(matches,
+ new TermsAggregate("properties.lastName"),
Profile.ITEM_TYPE);
+ Assert.assertEquals(Long.valueOf(count),
nonOptimized.get("_filtered"));
+
Assert.assertEquals(Long.valueOf(persistenceService.getAllItemsCount(Profile.ITEM_TYPE)),
nonOptimized.get("_all"));
+ } finally {
+ persistenceService.removeByQuery(matches, Profile.class);
+ refreshPersistence(Profile.class);
+ }
+ }
+
private Condition matchAllCondition() {
return new
Condition(definitionsService.getConditionType("matchAllCondition"));
}
diff --git
a/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ElasticSearchPersistenceServiceImpl.java
b/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ElasticSearchPersistenceServiceImpl.java
index b27807fec..ec7ffd2ad 100644
---
a/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ElasticSearchPersistenceServiceImpl.java
+++
b/persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ElasticSearchPersistenceServiceImpl.java
@@ -2507,12 +2507,12 @@ public class ElasticSearchPersistenceServiceImpl
implements PersistenceService,
Query tenantScopedQuery =
wrapWithTenantAndItemTypeQuery(itemType, Query.of(q -> q.matchAll(m -> m)),
finalTenantId);
- // Optimized: put the optional condition in the query and skip
the slow global aggregation.
- // Non-optimized: keep a filter aggregation for _filtered, but
always scope the query to the
- // current tenant. A global aggregation ignores the query and
would count every tenant's documents.
- // Also register aggregations under stable names ("filter",
"buckets", "missing"): the previous
- // code used String.valueOf(Kind) ("Global") while reading
Kind.jsonValue() ("global"), so POST
- // aggregates with a condition returned nothing (UNOMI-1014).
+ // Optimized: the optional condition goes into the query, so
_filtered is the hit count.
+ // Non-optimized: the query is the whole tenant (_all is the
hit count) and a filter aggregation
+ // gives _filtered. Never use a global aggregation: it ignores
the query and counts every tenant.
+ // hits.total stops at 10,000 unless hits are counted exactly;
aggregations visit every matching
+ // document anyway, so exact counting costs little.
+ searchSourceBuilder.trackTotalHits(t -> t.enabled(true));
if (optimizedQuery) {
if (filter != null) {
searchSourceBuilder.query(wrapWithTenantAndItemTypeQuery(itemType,
@@ -2523,8 +2523,6 @@ public class ElasticSearchPersistenceServiceImpl
implements PersistenceService,
searchSourceBuilder.aggregations(aggregationsByType);
} else {
searchSourceBuilder.query(tenantScopedQuery);
- // _all comes from hits.total, which is capped at 10,000
unless hits are tracked exactly.
- searchSourceBuilder.trackTotalHits(t -> t.enabled(true));
if (filter != null) {
Aggregation filterAggregation = new
Aggregation.Builder().filter(wrapWithTenantAndItemTypeQuery(
itemType,
conditionESQueryBuilderDispatcher.buildFilter(filter), finalTenantId))
@@ -2586,20 +2584,15 @@ public class ElasticSearchPersistenceServiceImpl
implements PersistenceService,
}
}
- long totalDocCount = addBucketCounts(bucketsAggregate,
aggregate, results);
+ addBucketCounts(bucketsAggregate, aggregate, results);
Aggregate missingAgg = aggregations.get("missing");
if (missingAgg != null && missingAgg.isMissing()) {
MissingAggregate missing = missingAgg.missing();
if (missing.docCount() > 0) {
results.put("_missing", missing.docCount());
- totalDocCount += missing.docCount();
}
}
- if (response.hits() != null && response.hits().total()
!= null
- &&
TotalHitsRelation.Gte.equals(response.hits().total().relation())) {
- results.put("_filtered", totalDocCount);
- }
}
}
return results;
diff --git
a/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/OpenSearchPersistenceServiceImpl.java
b/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/OpenSearchPersistenceServiceImpl.java
index 235818bd2..fb2ff5b5b 100644
---
a/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/OpenSearchPersistenceServiceImpl.java
+++
b/persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/OpenSearchPersistenceServiceImpl.java
@@ -2505,13 +2505,12 @@ public class OpenSearchPersistenceServiceImpl
implements PersistenceService, Syn
}
}
- // Optimized: put the optional condition in the query and skip
the slow global aggregation.
- // Non-optimized: keep a filter aggregation for _filtered,
scoped to the current tenant.
- // Register aggregations under stable names ("filter",
"buckets", "missing"). The previous
- // non-optimized path stored the global aggregation as
"buckets" while reading "global", and
- // nested the condition filter under "buckets" while reading
"filter", so POST aggregates with
- // a condition returned nothing (UNOMI-1014). A global
aggregation also ignores the query and
- // would count every tenant's documents.
+ // Optimized: the optional condition goes into the query, so
_filtered is the hit count.
+ // Non-optimized: the query is the whole tenant (_all is the
hit count) and a filter aggregation
+ // gives _filtered. Never use a global aggregation: it ignores
the query and counts every tenant.
+ // hits.total stops at 10,000 unless hits are counted exactly;
aggregations visit every matching
+ // document anyway, so exact counting costs little.
+ searchRequestBuilder.trackTotalHits(t -> t.enabled(true));
if (optimizedQuery) {
if (filter != null) {
searchRequestBuilder.query(wrapWithTenantAndItemTypeQuery(itemType,
@@ -2524,8 +2523,6 @@ public class OpenSearchPersistenceServiceImpl implements
PersistenceService, Syn
}
} else {
searchRequestBuilder.query(tenantScopedQuery);
- // _all comes from hits.total, which is capped at 10,000
unless hits are tracked exactly.
- searchRequestBuilder.trackTotalHits(t -> t.enabled(true));
if (filter != null) {
Aggregation.Builder.ContainerBuilder
filterAggregationContainerBuilder = new Aggregation.Builder().filter(
wrapWithTenantAndItemTypeQuery(itemType,
conditionOSQueryBuilderDispatcher.buildFilter(filter),
@@ -2585,19 +2582,14 @@ public class OpenSearchPersistenceServiceImpl
implements PersistenceService, Syn
}
}
- long totalDocCount = addBucketCounts(bucketsAggregate,
aggregate, results);
+ addBucketCounts(bucketsAggregate, aggregate, results);
Aggregate missingAgg = aggregates.get("missing");
if (missingAgg != null && missingAgg.isMissing()) {
MissingAggregate missing = missingAgg.missing();
if (missing.docCount() > 0) {
results.put("_missing", missing.docCount());
- totalDocCount += missing.docCount();
}
}
- if (response.hits() != null && response.hits().total()
!= null
- &&
TotalHitsRelation.Gte.equals(response.hits().total().relation())) {
- results.put("_filtered", totalDocCount);
- }
}
}
return results;
diff --git
a/persistence-spi/src/main/java/org/apache/unomi/persistence/spi/PersistenceService.java
b/persistence-spi/src/main/java/org/apache/unomi/persistence/spi/PersistenceService.java
index 430e71ac8..55d9f7c10 100644
---
a/persistence-spi/src/main/java/org/apache/unomi/persistence/spi/PersistenceService.java
+++
b/persistence-spi/src/main/java/org/apache/unomi/persistence/spi/PersistenceService.java
@@ -677,7 +677,8 @@ public interface PersistenceService {
/**
* Aggregates item counts for the given type, optionally filtered by
condition and
* aggregated according to the specified {@link BaseAggregate}.
- * This aggregate won't return the {@code _all} count and should therefore
be much faster than {@link #aggregateQuery(Condition, BaseAggregate, String)}
+ * Unlike {@link #aggregateQuery(Condition, BaseAggregate, String)}, it
does not return the {@code _all} count, which saves
+ * counting every item of the type.
*
* @param filter the condition the items must match or {@code null} if
no filtering is needed
* @param aggregate an aggregate specifying how matching items must be
bundled
diff --git
a/rest/src/main/java/org/apache/unomi/rest/endpoints/QueryServiceEndPoint.java
b/rest/src/main/java/org/apache/unomi/rest/endpoints/QueryServiceEndPoint.java
index 6a6d3c058..36b5a5fa3 100644
---
a/rest/src/main/java/org/apache/unomi/rest/endpoints/QueryServiceEndPoint.java
+++
b/rest/src/main/java/org/apache/unomi/rest/endpoints/QueryServiceEndPoint.java
@@ -92,7 +92,7 @@ public class QueryServiceEndPoint {
/**
* Returns property-value counts for an item type, optionally using an
aggregate query.
* <p>
- * When {@code optimizedQuery} is {@code true}, the {@code _all} count of
the tenant's items of that type is omitted for faster execution.
+ * When {@code optimizedQuery} is {@code true}, the {@code _all} count of
the tenant's items of that type is omitted, which saves counting them.
*
* @param type the item type name from the class {@code ITEM_TYPE} field
* @param property the property whose distinct values form aggregation
buckets
