This is an automated email from the ASF dual-hosted git repository. sergehuber pushed a commit to branch UNOMI-1014-aggregate-tenant-filter in repository https://gitbox.apache.org/repos/asf/unomi.git
commit 6872c08c73e900c907a06955c2f17da8fd0a7705 Author: Serge Huber <[email protected]> AuthorDate: Fri Oct 23 09:41:55 2026 +0200 UNOMI-1014: Return aggregate results and keep them in the current tenant POST aggregates failed because Elasticsearch and OpenSearch used mismatched aggregation names when reading the response. Aggregates and metrics now always run in the current tenant, including the non-optimized path that previously used a global aggregation. --- .../apache/unomi/itests/PersistenceServiceIT.java | 119 +++++++++++++++++++++ .../ElasticSearchPersistenceServiceImpl.java | 77 ++++++++----- .../OpenSearchPersistenceServiceImpl.java | 85 +++++++++------ 3 files changed, 220 insertions(+), 61 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 05d00aef4..fdbd2b296 100644 --- a/itests/src/test/java/org/apache/unomi/itests/PersistenceServiceIT.java +++ b/itests/src/test/java/org/apache/unomi/itests/PersistenceServiceIT.java @@ -19,7 +19,9 @@ package org.apache.unomi.itests; import org.apache.unomi.api.Event; import org.apache.unomi.api.PartialList; import org.apache.unomi.api.Profile; +import org.apache.unomi.api.conditions.Condition; import org.apache.unomi.api.tenants.Tenant; +import org.apache.unomi.persistence.spi.aggregate.TermsAggregate; import org.junit.After; import org.junit.Assert; import org.junit.Test; @@ -32,6 +34,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.Date; import java.util.List; +import java.util.Map; import java.util.UUID; /** @@ -213,6 +216,122 @@ public class PersistenceServiceIT extends BaseIT { } } + /** + * UNOMI-1014: a terms aggregate with a condition must return buckets again (non-optimized and optimized). + */ + @Test + public void testAggregateWithConditionReturnsBuckets() throws InterruptedException { + String suffix = UUID.randomUUID().toString(); + saveProfileWithFirstName("agg-alice-" + suffix, "Alice"); + saveProfileWithFirstName("agg-alice-2-" + suffix, "Alice"); + saveProfileWithFirstName("agg-bob-" + suffix, "Bob"); + refreshPersistence(Profile.class); + + Condition matchAll = matchAllCondition(); + TermsAggregate byFirstName = new TermsAggregate("properties.firstName"); + + Map<String, Long> withCondition = keepTrying( + "Aggregate with a condition should return first-name buckets", + () -> persistenceService.aggregateQuery(matchAll, byFirstName, Profile.ITEM_TYPE), + r -> r != null && Long.valueOf(2L).equals(r.get("Alice")) && Long.valueOf(1L).equals(r.get("Bob")), + DEFAULT_TRYING_TIMEOUT, DEFAULT_TRYING_TRIES); + Assert.assertEquals(Long.valueOf(2L), withCondition.get("Alice")); + Assert.assertEquals(Long.valueOf(1L), withCondition.get("Bob")); + Assert.assertTrue("Non-optimized aggregate should report a tenant-scoped _all count", + withCondition.get("_all") != null && withCondition.get("_all") >= 3L); + Assert.assertTrue("Non-optimized aggregate with a condition should report _filtered", + withCondition.get("_filtered") != null && withCondition.get("_filtered") >= 3L); + + Map<String, Long> optimized = keepTrying( + "Optimized aggregate with a condition should return first-name buckets", + () -> persistenceService.aggregateWithOptimizedQuery(matchAll, byFirstName, Profile.ITEM_TYPE), + r -> r != null && Long.valueOf(2L).equals(r.get("Alice")) && Long.valueOf(1L).equals(r.get("Bob")), + DEFAULT_TRYING_TIMEOUT, DEFAULT_TRYING_TRIES); + Assert.assertEquals(Long.valueOf(2L), optimized.get("Alice")); + Assert.assertEquals(Long.valueOf(1L), optimized.get("Bob")); + } + + /** + * Aggregate and metric queries must only see the current tenant's documents. + */ + @Test + public void testAggregateAndMetricsStayInTheCurrentTenant() throws Exception { + String suffix = UUID.randomUUID().toString().substring(0, 8); + String localAlice = "agg-tenant-local-alice-" + suffix; + String otherAlice = "agg-tenant-other-alice-" + suffix; + String otherBob = "agg-tenant-other-bob-" + suffix; + Tenant otherTenant = tenantService.createTenant("agg-other-" + suffix, Collections.emptyMap()); + try { + saveProfileWithFirstName(localAlice, "Alice"); + Profile local = persistenceService.load(localAlice, Profile.class); + local.setProperty("age", 7); + persistenceService.save(local); + executionContextManager.executeAsTenant(otherTenant.getItemId(), () -> { + Profile otherAliceProfile = new Profile(otherAlice); + otherAliceProfile.setProperty("firstName", "Alice"); + otherAliceProfile.setProperty("age", 10); + persistenceService.save(otherAliceProfile); + Profile otherBobProfile = new Profile(otherBob); + otherBobProfile.setProperty("firstName", "Bob"); + otherBobProfile.setProperty("age", 5); + persistenceService.save(otherBobProfile); + }); + refreshPersistence(Profile.class); + + Condition matchAll = matchAllCondition(); + TermsAggregate byFirstName = new TermsAggregate("properties.firstName"); + + Map<String, Long> termsNoCondition = keepTrying( + "Terms aggregate without a condition should only count the current tenant", + () -> persistenceService.aggregateWithOptimizedQuery(null, byFirstName, Profile.ITEM_TYPE), + r -> r != null && Long.valueOf(1L).equals(r.get("Alice")) && !r.containsKey("Bob"), + DEFAULT_TRYING_TIMEOUT, DEFAULT_TRYING_TRIES); + Assert.assertEquals(Long.valueOf(1L), termsNoCondition.get("Alice")); + Assert.assertNull("Another tenant's first names must not appear", termsNoCondition.get("Bob")); + + Map<String, Long> termsWithCondition = keepTrying( + "Terms aggregate with a condition should only count the current tenant", + () -> persistenceService.aggregateQuery(matchAll, byFirstName, Profile.ITEM_TYPE), + r -> r != null && Long.valueOf(1L).equals(r.get("Alice")) && !r.containsKey("Bob"), + DEFAULT_TRYING_TIMEOUT, DEFAULT_TRYING_TRIES); + Assert.assertEquals(Long.valueOf(1L), termsWithCondition.get("Alice")); + Assert.assertNull(termsWithCondition.get("Bob")); + Assert.assertNotNull("Non-optimized aggregate should report _all for the current tenant", + termsWithCondition.get("_all")); + Assert.assertEquals("With match-all, _all and _filtered must match for the current tenant", + termsWithCondition.get("_all"), termsWithCondition.get("_filtered")); + + Map<String, Double> metrics = keepTrying( + "Sum metric should only include the current tenant", + () -> persistenceService.getSingleValuesMetrics(matchAll, new String[]{"sum", "count"}, + AGE_PROPERTY, Profile.ITEM_TYPE), + r -> r != null && r.get("_sum") != null && r.get("_sum") == 7.0, + DEFAULT_TRYING_TIMEOUT, DEFAULT_TRYING_TRIES); + Assert.assertEquals(7.0, metrics.get("_sum"), 0.001); + Assert.assertEquals(1.0, metrics.get("_count"), 0.001); + } finally { + try { + executionContextManager.executeAsTenant(otherTenant.getItemId(), () -> { + persistenceService.remove(otherAlice, Profile.class); + persistenceService.remove(otherBob, Profile.class); + }); + } finally { + tenantService.deleteTenant(otherTenant.getItemId()); + } + } + } + + private Condition matchAllCondition() { + return new Condition(definitionsService.getConditionType("matchAllCondition")); + } + + private void saveProfileWithFirstName(String itemId, String firstName) { + Profile profile = new Profile(itemId); + profile.setProperty("firstName", firstName); + persistenceService.save(profile); + profileIds.add(itemId); + } + private Profile saveProfile(String itemId) { Profile profile = new Profile(itemId); persistenceService.save(profile); 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 46cdad5ba..14a00678d 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 @@ -2501,28 +2501,35 @@ public class ElasticSearchPersistenceServiceImpl implements PersistenceService, SearchRequest.Builder searchSourceBuilder = new SearchRequest.Builder(); searchSourceBuilder.index(getIndexNameForQuery(itemType)); searchSourceBuilder.size(0); - searchSourceBuilder.query( - isItemTypeSharingIndex(itemType) ? getItemTypeQuery(itemType) : Query.of(q -> q.matchAll(m -> m))); - // If the request is optimized then we don't need a global aggregation which is very slow and we can put the query with a - // filter on range items in the query block so we don't retrieve all the document before filtering the whole - if (optimizedQuery) { - searchSourceBuilder.aggregations(aggregationsByType); + Query itemTypeOrMatchAll = + isItemTypeSharingIndex(itemType) ? getItemTypeQuery(itemType) : Query.of(q -> q.matchAll(m -> m)); + Query tenantScopedQuery = wrapWithTenantAndItemTypeQuery(itemType, itemTypeOrMatchAll, 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). + if (optimizedQuery) { if (filter != null) { - searchSourceBuilder.query(wrapWithTenantAndItemTypeQuery(itemType, conditionESQueryBuilderDispatcher.buildFilter(filter), finalTenantId)); + searchSourceBuilder.query(wrapWithTenantAndItemTypeQuery(itemType, + conditionESQueryBuilderDispatcher.buildFilter(filter), finalTenantId)); + } else { + searchSourceBuilder.query(tenantScopedQuery); } + searchSourceBuilder.aggregations(aggregationsByType); } else { + searchSourceBuilder.query(tenantScopedQuery); if (filter != null) { - Aggregation.Builder aggBuilder = new Aggregation.Builder(); - aggBuilder.filter(wrapWithTenantAndItemTypeQuery(itemType, conditionESQueryBuilderDispatcher.buildFilter(filter), finalTenantId)) - .aggregations(aggregationsByType); - - aggregationsByType = Map.of("filter", aggBuilder.build()); + Aggregation filterAggregation = new Aggregation.Builder().filter(wrapWithTenantAndItemTypeQuery( + itemType, conditionESQueryBuilderDispatcher.buildFilter(filter), finalTenantId)) + .aggregations(aggregationsByType).build(); + searchSourceBuilder.aggregations("filter", filterAggregation); + } else { + searchSourceBuilder.aggregations(aggregationsByType); } - Aggregation globalAgg = new Aggregation.Builder().global(new GlobalAggregation.Builder().build()) - .aggregations(aggregationsByType).build(); - searchSourceBuilder.aggregations(String.valueOf(globalAgg._kind()), globalAgg); } RequestOptions.Builder builder = RequestOptions.DEFAULT.toBuilder(); @@ -2550,16 +2557,17 @@ public class ElasticSearchPersistenceServiceImpl implements PersistenceService, if (aggregations != null) { if (optimizedQuery) { - if (response.hits() != null) { + if (response.hits() != null && response.hits().total() != null) { results.put("_filtered", response.hits().total().value()); } } else { - GlobalAggregate globalAggregate = aggregations.get(Aggregation.Kind.Global.jsonValue()).global(); - results.put("_all", globalAggregate.docCount()); - aggregations = globalAggregate.aggregations(); + if (response.hits() != null && response.hits().total() != null) { + results.put("_all", response.hits().total().value()); + } - if (aggregations.get(Aggregate.Kind.Filter.jsonValue()) != null) { - FilterAggregate filterAggregate = aggregations.get(Aggregation.Kind.Filter.jsonValue()).filter(); + Aggregate filterAggResult = aggregations.get("filter"); + if (filterAggResult != null && filterAggResult.isFilter()) { + FilterAggregate filterAggregate = filterAggResult.filter(); results.put("_filtered", filterAggregate.docCount()); aggregations = filterAggregate.aggregations(); } @@ -2605,12 +2613,16 @@ public class ElasticSearchPersistenceServiceImpl implements PersistenceService, } } - MissingAggregate missing = aggregations.get("missing").missing(); - if (missing.docCount() > 0) { - results.put("_missing", missing.docCount()); - totalDocCount += missing.docCount(); + 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 && TotalHitsRelation.Gte.equals(response.hits().total().relation())) { + if (response.hits() != null && response.hits().total() != null + && TotalHitsRelation.Gte.equals(response.hits().total().relation())) { results.put("_filtered", totalDocCount); } } @@ -2802,6 +2814,7 @@ public class ElasticSearchPersistenceServiceImpl implements PersistenceService, @Override public Map<String, Double> getSingleValuesMetrics(final Condition condition, final String[] metrics, final String field, final String itemType) { + String finalTenantId = validateTenantAndGetId(SecurityServiceConfiguration.PERMISSION_AGGREGATE); return new InClassLoaderExecute<Map<String, Double>>(metricsService, this.getClass().getName() + ".getSingleValuesMetrics", this.bundleContext, this.fatalIllegalStateErrors, throwExceptions) { @@ -2834,11 +2847,17 @@ public class ElasticSearchPersistenceServiceImpl implements PersistenceService, } } - Aggregation filterAggregation = Aggregation.of( - a -> a.filter(conditionESQueryBuilderDispatcher.buildFilter(condition)).aggregations(subAggs)); + Query conditionQuery = condition != null ? + conditionESQueryBuilderDispatcher.buildFilter(condition) : + Query.of(q -> q.matchAll(m -> m)); + Query itemTypeOrMatchAll = + isItemTypeSharingIndex(itemType) ? getItemTypeQuery(itemType) : Query.of(q -> q.matchAll(m -> m)); + Aggregation filterAggregation = Aggregation.of(a -> a.filter( + wrapWithTenantAndItemTypeQuery(itemType, conditionQuery, finalTenantId)) + .aggregations(subAggs)); SearchRequest searchRequest = SearchRequest.of( s -> s.index(getIndexNameForQuery(itemType)).size(0).source(builder -> builder.fetch(true)) - .query(isItemTypeSharingIndex(itemType) ? getItemTypeQuery(itemType) : Query.of(q -> q.matchAll(m -> m))) + .query(wrapWithTenantAndItemTypeQuery(itemType, itemTypeOrMatchAll, finalTenantId)) .aggregations("metrics", filterAggregation)); SearchResponse<Void> searchResponse = esClient.search(searchRequest); 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 a83d6a7b6..8b5d3a6df 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 @@ -2404,8 +2404,8 @@ public class OpenSearchPersistenceServiceImpl implements PersistenceService, Syn SearchRequest.Builder searchRequestBuilder = new SearchRequest.Builder().index(getIndexNameForQuery(itemType)); searchRequestBuilder.size(0); Query matchAll = Query.of(q->q.matchAll(m->m)); - boolean isItemTypeSharingIndex = isItemTypeSharingIndex(itemType); - searchRequestBuilder.query(wrapWithTenantAndItemTypeQuery(itemType,matchAll, finalTenantId)); + Query tenantScopedQuery = wrapWithTenantAndItemTypeQuery(itemType, matchAll, finalTenantId); + searchRequestBuilder.query(tenantScopedQuery); Map<String, Aggregation.Builder.ContainerBuilder> lastAggregation = new LinkedHashMap<>(); if (aggregate != null) { @@ -2495,31 +2495,39 @@ public class OpenSearchPersistenceServiceImpl implements PersistenceService, Syn } } - // If the request is optimized then we don't need a global aggregation which is very slow and we can put the query with a - // filter on range items in the query block so we don't retrieve all the document before filtering the whole + // 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. if (optimizedQuery) { + if (filter != null) { + searchRequestBuilder.query(wrapWithTenantAndItemTypeQuery(itemType, + conditionOSQueryBuilderDispatcher.buildFilter(filter), finalTenantId)); + } else { + searchRequestBuilder.query(tenantScopedQuery); + } for (Map.Entry<String, Aggregation.Builder.ContainerBuilder> aggregationBuilder : lastAggregation.entrySet()) { searchRequestBuilder.aggregations(aggregationBuilder.getKey(), aggregationBuilder.getValue().build()); } - - if (filter != null) { - searchRequestBuilder.query(wrapWithTenantAndItemTypeQuery(itemType, conditionOSQueryBuilderDispatcher.buildFilter(filter), finalTenantId)); - } } else { + searchRequestBuilder.query(tenantScopedQuery); if (filter != null) { - Aggregation.Builder.ContainerBuilder filterAggregationContainerBuilder = new Aggregation.Builder().filter(wrapWithTenantAndItemTypeQuery(itemType, conditionOSQueryBuilderDispatcher.buildFilter(filter), finalTenantId)); - for (Map.Entry<String,Aggregation.Builder.ContainerBuilder> aggregationBuilder : lastAggregation.entrySet()) { - filterAggregationContainerBuilder.aggregations(aggregationBuilder.getKey(), aggregationBuilder.getValue().build()); + Aggregation.Builder.ContainerBuilder filterAggregationContainerBuilder = new Aggregation.Builder().filter( + wrapWithTenantAndItemTypeQuery(itemType, conditionOSQueryBuilderDispatcher.buildFilter(filter), + finalTenantId)); + for (Map.Entry<String, Aggregation.Builder.ContainerBuilder> aggregationBuilder : lastAggregation.entrySet()) { + filterAggregationContainerBuilder.aggregations(aggregationBuilder.getKey(), + aggregationBuilder.getValue().build()); + } + searchRequestBuilder.aggregations("filter", filterAggregationContainerBuilder.build()); + } else { + for (Map.Entry<String, Aggregation.Builder.ContainerBuilder> aggregationBuilder : lastAggregation.entrySet()) { + searchRequestBuilder.aggregations(aggregationBuilder.getKey(), aggregationBuilder.getValue().build()); } - lastAggregation = Map.of("buckets", filterAggregationContainerBuilder); - } - - Aggregation.Builder.ContainerBuilder globalAggregationContainerBuilder = new Aggregation.Builder().global(g -> g.name("global")); - for (Map.Entry<String, Aggregation.Builder.ContainerBuilder> aggregationBuilder : lastAggregation.entrySet()) { - globalAggregationContainerBuilder.aggregations(aggregationBuilder.getKey(), aggregationBuilder.getValue().build()); } - - searchRequestBuilder.aggregations("buckets", globalAggregationContainerBuilder.build()); } /* @@ -2540,16 +2548,17 @@ public class OpenSearchPersistenceServiceImpl implements PersistenceService, Syn if (aggregates != null) { if (optimizedQuery) { - if (response.hits() != null) { + if (response.hits() != null && response.hits().total() != null) { results.put("_filtered", response.hits().total().value()); } } else { - GlobalAggregate globalAgg = aggregates.get("global").global(); - results.put("_all", globalAgg.docCount()); - aggregates = globalAgg.aggregations(); + if (response.hits() != null && response.hits().total() != null) { + results.put("_all", response.hits().total().value()); + } - if (aggregates.get("filter") != null) { - FilterAggregate filterAgg = aggregates.get("filter").filter(); + Aggregate filterAggResult = aggregates.get("filter"); + if (filterAggResult != null && filterAggResult.isFilter()) { + FilterAggregate filterAgg = filterAggResult.filter(); results.put("_filtered", filterAgg.docCount()); aggregates = filterAgg.aggregations(); } @@ -2574,12 +2583,16 @@ public class OpenSearchPersistenceServiceImpl implements PersistenceService, Syn totalDocCount += bucket.docCount(); } } - MissingAggregate missing = aggregates.get("missing").missing(); - if (missing.docCount() > 0) { - results.put("_missing", missing.docCount()); - totalDocCount += missing.docCount(); + 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 && TotalHitsRelation.Gte.equals(response.hits().total().relation())) { + if (response.hits() != null && response.hits().total() != null + && TotalHitsRelation.Gte.equals(response.hits().total().relation())) { results.put("_filtered", totalDocCount); } } @@ -2742,13 +2755,21 @@ public class OpenSearchPersistenceServiceImpl implements PersistenceService, Syn @Override public Map<String, Double> getSingleValuesMetrics(final Condition condition, final String[] metrics, final String field, final String itemType) { + String finalTenantId = validateTenantAndGetId(SecurityServiceConfiguration.PERMISSION_AGGREGATE); return new InClassLoaderExecute<Map<String, Double>>(metricsService, this.getClass().getName() + ".getSingleValuesMetrics", this.bundleContext, this.fatalIllegalStateErrors, throwExceptions) { @Override protected Map<String, Double> execute(Object... args) throws IOException { Map<String, Double> results = new LinkedHashMap<>(); - Aggregation.Builder.ContainerBuilder filterAggregationContainerBuilder = new Aggregation.Builder().filter(conditionOSQueryBuilderDispatcher.buildFilter(condition)); + Query conditionQuery = condition != null ? + conditionOSQueryBuilderDispatcher.buildFilter(condition) : + Query.of(q -> q.matchAll(i -> i)); + Query itemTypeOrMatchAll = isItemTypeSharingIndex(itemType) ? + getItemTypeQueryBuilder(itemType) : + Query.of(q -> q.matchAll(i -> i)); + Aggregation.Builder.ContainerBuilder filterAggregationContainerBuilder = new Aggregation.Builder().filter( + wrapWithTenantAndItemTypeQuery(itemType, conditionQuery, finalTenantId)); if (metrics != null) { for (String metric : metrics) { @@ -2778,7 +2799,7 @@ public class OpenSearchPersistenceServiceImpl implements PersistenceService, Syn .index(getIndexNameForQuery(itemType)) .size(0) .aggregations("metrics", filterAggregationContainerBuilder.build()) - .query(isItemTypeSharingIndex(itemType) ? getItemTypeQueryBuilder(itemType) : Query.of(q -> q.matchAll(i -> i))) + .query(wrapWithTenantAndItemTypeQuery(itemType, itemTypeOrMatchAll, finalTenantId)) , Item.class );
