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 71b073534 UNOMI-1014: Return aggregate results and keep them in the
current tenant (#922)
71b073534 is described below
commit 71b0735344ec0e7dcd9c5ce15360efab290e0f9b
Author: Serge Huber <[email protected]>
AuthorDate: Fri Oct 23 16:36:35 2026 +0200
UNOMI-1014: Return aggregate results and keep them in the current tenant
(#922)
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../apache/unomi/itests/PersistenceServiceIT.java | 119 +++++++++++++++++++++
.../ElasticSearchPersistenceServiceImpl.java | 75 ++++++++-----
.../OpenSearchPersistenceServiceImpl.java | 84 +++++++++------
3 files changed, 217 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..f5980a7ac 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 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).
+ 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);
+ // _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.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,15 @@ 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));
+ 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, Query.of(q -> q.matchAll(m ->
m)), 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..0e2be7a7e 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,41 @@ 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);
+ // _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), 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 +2550,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 +2585,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 +2757,18 @@ 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));
+ Aggregation.Builder.ContainerBuilder
filterAggregationContainerBuilder = new Aggregation.Builder().filter(
+ wrapWithTenantAndItemTypeQuery(itemType,
conditionQuery, finalTenantId));
if (metrics != null) {
for (String metric : metrics) {
@@ -2778,7 +2798,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,
Query.of(q -> q.matchAll(i -> i)), finalTenantId))
, Item.class
);
