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 b5e1bddbc UNOMI-1018: Return date and range buckets from aggregate
queries (#926)
b5e1bddbc is described below
commit b5e1bddbc6dc3be9073db9c2e05fa7bf43436cd9
Author: Serge Huber <[email protected]>
AuthorDate: Mon Oct 26 10:16:32 2026 +0100
UNOMI-1018: Return date and range buckets from aggregate queries (#926)
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../java/org/apache/unomi/api/query/DateRange.java | 6 +-
.../java/org/apache/unomi/api/query/IpRange.java | 9 +-
.../org/apache/unomi/api/query/NumericRange.java | 6 +-
.../apache/unomi/itests/PersistenceServiceIT.java | 138 ++++++++++++++++-
.../ElasticSearchPersistenceServiceImpl.java | 103 ++++++++-----
.../ElasticSearchAggregateBucketsTest.java | 159 +++++++++++++++++++
.../OpenSearchPersistenceServiceImpl.java | 115 ++++++++++----
.../opensearch/OpenSearchAggregateBucketsTest.java | 152 ++++++++++++++++++
.../spi/aggregate/AggregateBuckets.java | 171 +++++++++++++++++++++
.../spi/aggregate/AggregateBucketsTest.java | 154 +++++++++++++++++++
10 files changed, 930 insertions(+), 83 deletions(-)
diff --git a/api/src/main/java/org/apache/unomi/api/query/DateRange.java
b/api/src/main/java/org/apache/unomi/api/query/DateRange.java
index cfdfb01e5..e8c5066b0 100644
--- a/api/src/main/java/org/apache/unomi/api/query/DateRange.java
+++ b/api/src/main/java/org/apache/unomi/api/query/DateRange.java
@@ -20,7 +20,7 @@ package org.apache.unomi.api.query;
import java.io.Serializable;
/**
- * Named inclusive date bucket for queries and aggregations.
+ * Named date bucket for queries and aggregations.
* Combines a display {@code key} with {@code from} and {@code to} bounds so
* segment conditions and aggregate queries can group events or profiles inside
* a calendar window.
@@ -73,7 +73,7 @@ public class DateRange implements Serializable {
}
/**
- * Inclusive upper bound of the interval.
+ * Upper bound of the interval, exclusive in range aggregations.
*
* @return end date or value
*/
@@ -82,7 +82,7 @@ public class DateRange implements Serializable {
}
/**
- * Sets the inclusive upper bound.
+ * Sets the upper bound, exclusive in range aggregations.
*
* @param to end date or value
*/
diff --git a/api/src/main/java/org/apache/unomi/api/query/IpRange.java
b/api/src/main/java/org/apache/unomi/api/query/IpRange.java
index a2deb51ed..2e9121b58 100644
--- a/api/src/main/java/org/apache/unomi/api/query/IpRange.java
+++ b/api/src/main/java/org/apache/unomi/api/query/IpRange.java
@@ -20,9 +20,8 @@ package org.apache.unomi.api.query;
import java.io.Serializable;
/**
- * Named inclusive IP range used in geo or network segment conditions.
- * The {@code from} and {@code to} strings define lower and upper bounds
checked
- * when evaluating whether a visitor's IP address belongs to the range.
+ * Named IP range used by property types and IP range aggregations.
+ * In range aggregations, {@code from} is inclusive and {@code to} is
exclusive.
*/
public class IpRange implements Serializable {
private String key;
@@ -72,7 +71,7 @@ public class IpRange implements Serializable {
}
/**
- * Inclusive upper IP bound.
+ * Upper IP bound, exclusive in range aggregations.
*
* @return end IP address
*/
@@ -81,7 +80,7 @@ public class IpRange implements Serializable {
}
/**
- * Sets the inclusive upper IP bound.
+ * Sets the upper IP bound, exclusive in range aggregations.
*
* @param to end IP address
*/
diff --git a/api/src/main/java/org/apache/unomi/api/query/NumericRange.java
b/api/src/main/java/org/apache/unomi/api/query/NumericRange.java
index 9a493b79b..3fae05f5c 100644
--- a/api/src/main/java/org/apache/unomi/api/query/NumericRange.java
+++ b/api/src/main/java/org/apache/unomi/api/query/NumericRange.java
@@ -20,7 +20,7 @@ package org.apache.unomi.api.query;
import java.io.Serializable;
/**
- * Named inclusive numeric bucket for query filters and aggregations.
+ * Named numeric bucket for query filters and aggregations.
* Stores a {@code key} plus {@code from}/{@code to} bounds so conditions can
* test profile or event properties against numeric intervals.
*/
@@ -72,7 +72,7 @@ public class NumericRange implements Serializable {
}
/**
- * Inclusive upper bound.
+ * Upper bound, exclusive in range aggregations.
*
* @return upper bound, or {@code null} if unset
*/
@@ -81,7 +81,7 @@ public class NumericRange implements Serializable {
}
/**
- * Sets the inclusive upper bound.
+ * Sets the upper bound, exclusive in range aggregations.
*
* @param to upper bound
*/
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 eb039b203..75ac30fe7 100644
--- a/itests/src/test/java/org/apache/unomi/itests/PersistenceServiceIT.java
+++ b/itests/src/test/java/org/apache/unomi/itests/PersistenceServiceIT.java
@@ -20,7 +20,12 @@ 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.query.DateRange;
+import org.apache.unomi.api.query.NumericRange;
import org.apache.unomi.api.tenants.Tenant;
+import org.apache.unomi.persistence.spi.aggregate.DateAggregate;
+import org.apache.unomi.persistence.spi.aggregate.DateRangeAggregate;
+import org.apache.unomi.persistence.spi.aggregate.NumericRangeAggregate;
import org.apache.unomi.persistence.spi.aggregate.TermsAggregate;
import org.junit.After;
import org.junit.Assert;
@@ -30,7 +35,9 @@ import org.ops4j.pax.exam.junit.PaxExam;
import org.ops4j.pax.exam.spi.reactors.ExamReactorStrategy;
import org.ops4j.pax.exam.spi.reactors.PerSuite;
+import java.time.Instant;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.List;
@@ -312,7 +319,7 @@ public class PersistenceServiceIT extends BaseIT {
Assert.assertEquals(7.0, metrics.get("_sum"), 0.001);
Assert.assertEquals(1.0, metrics.get("_count"), 0.001);
- // Non-optimized path without a condition: this is where the
global aggregation used to count every tenant.
+ // Without a condition the non-optimized aggregate must still only
count the current tenant.
Map<String, Long> termsNonOptimizedNoCondition = keepTrying(
"Non-optimized terms aggregate without a condition should
only count the current tenant",
() -> persistenceService.aggregateQuery(null, byFirstName,
Profile.ITEM_TYPE),
@@ -332,6 +339,16 @@ public class PersistenceServiceIT extends BaseIT {
DEFAULT_TRYING_TIMEOUT, DEFAULT_TRYING_TRIES);
Assert.assertEquals(7.0, metricsNoCondition.get("_sum"), 0.001);
Assert.assertEquals(1.0, metricsNoCondition.get("_count"), 0.001);
+
+ Map<String, Long> otherTenantTerms = keepTrying(
+ "The other tenant's _all should only count its own
profiles",
+ () ->
executionContextManager.executeAsTenant(otherTenant.getItemId(),
+ () -> persistenceService.aggregateQuery(null,
byFirstName, Profile.ITEM_TYPE)),
+ r -> r != null && Long.valueOf(2L).equals(r.get("_all")),
+ DEFAULT_TRYING_TIMEOUT, DEFAULT_TRYING_TRIES);
+ Assert.assertEquals(Long.valueOf(2L),
otherTenantTerms.get("_all"));
+ Assert.assertEquals(Long.valueOf(1L),
otherTenantTerms.get("Alice"));
+ Assert.assertEquals(Long.valueOf(1L), otherTenantTerms.get("Bob"));
} finally {
try {
executionContextManager.executeAsTenant(otherTenant.getItemId(), () -> {
@@ -344,10 +361,129 @@ public class PersistenceServiceIT extends BaseIT {
}
}
+ /**
+ * Date histogram, numeric terms, numeric range and date range aggregates
must return their buckets on both engines.
+ */
+ @Test
+ public void testDateAndRangeAggregatesReturnBuckets() throws
InterruptedException {
+ String marker = "agg-buckets-" + UUID.randomUUID();
+ saveProfileWithVisitAndAge(marker, "2020-01-01T10:00:00Z", 10);
+ saveProfileWithVisitAndAge(marker, "2020-01-01T11:00:00Z", 30);
+ saveProfileWithVisitAndAge(marker, "2020-01-01T12:00:00Z", 40);
+ saveProfileWithVisitAndAge(marker, "2020-01-03T10:00:00Z", 70);
+ refreshPersistence(Profile.class);
+ Condition onlyTheseProfiles = firstNameEquals(marker);
+
+ // "1d" is the legacy interval format; both engines must accept it
+ Map<String, Long> byDay = keepTrying("Date histogram should return one
bucket per day",
+ () ->
persistenceService.aggregateWithOptimizedQuery(onlyTheseProfiles,
+ new DateAggregate("properties.lastVisit", "1d",
"yyyy-MM-dd"), Profile.ITEM_TYPE),
+ r -> r != null && Long.valueOf(3L).equals(r.get("2020-01-01")),
+ DEFAULT_TRYING_TIMEOUT, DEFAULT_TRYING_TRIES);
+ Assert.assertEquals(Long.valueOf(3L), byDay.get("2020-01-01"));
+ Assert.assertEquals(Long.valueOf(0L), byDay.get("2020-01-02"));
+ Assert.assertEquals(Long.valueOf(1L), byDay.get("2020-01-03"));
+
+ Map<String, Long> byAgeValue = keepTrying("Terms aggregate on a
numeric field should key buckets by value",
+ () ->
persistenceService.aggregateWithOptimizedQuery(onlyTheseProfiles, new
TermsAggregate(AGE_PROPERTY),
+ Profile.ITEM_TYPE),
+ r -> r != null && Long.valueOf(1L).equals(r.get("70")),
+ DEFAULT_TRYING_TIMEOUT, DEFAULT_TRYING_TRIES);
+ for (String age : new String[]{"10", "30", "40", "70"}) {
+ Assert.assertEquals("Bucket for age " + age, Long.valueOf(1L),
byAgeValue.get(age));
+ }
+ Assert.assertFalse("No bucket may have a null key",
byAgeValue.containsKey(null));
+
+ Map<String, Long> byAge = keepTrying("Numeric range aggregate should
return one bucket per range",
+ () -> persistenceService.aggregateQuery(onlyTheseProfiles, new
NumericRangeAggregate(AGE_PROPERTY,
+ Arrays.asList(numericRange("young", null, 18.0),
numericRange("adult", 18.0, 65.0),
+ numericRange("senior", 65.0, null),
numericRange("all", null, null))), Profile.ITEM_TYPE),
+ r -> r != null && Long.valueOf(2L).equals(r.get("adult")),
+ DEFAULT_TRYING_TIMEOUT, DEFAULT_TRYING_TRIES);
+ Assert.assertEquals(Long.valueOf(1L), byAge.get("young"));
+ Assert.assertEquals(Long.valueOf(2L), byAge.get("adult"));
+ Assert.assertEquals(Long.valueOf(1L), byAge.get("senior"));
+ Assert.assertEquals("A range without bounds covers every profile",
Long.valueOf(4L), byAge.get("all"));
+ Assert.assertEquals(Long.valueOf(4L), byAge.get("_filtered"));
+
+ // One range has no lower bound and the other no upper bound; both
engines must accept them
+ Map<String, Long> byVisit = keepTrying("Date range aggregate should
return one bucket per range",
+ () ->
persistenceService.aggregateWithOptimizedQuery(onlyTheseProfiles, new
DateRangeAggregate(
+ "properties.lastVisit", "yyyy-MM-dd",
Arrays.asList(dateRange("before", null, "2020-01-02"),
+ dateRange("after", "2020-01-02", null))),
Profile.ITEM_TYPE),
+ r -> r != null && Long.valueOf(3L).equals(r.get("before")),
+ DEFAULT_TRYING_TIMEOUT, DEFAULT_TRYING_TRIES);
+ Assert.assertEquals(Long.valueOf(3L), byVisit.get("before"));
+ Assert.assertEquals(Long.valueOf(1L), byVisit.get("after"));
+ }
+
+ /**
+ * A bucket whose key is the name of a result count must not change that
count.
+ */
+ @Test
+ public void testBucketNamedLikeAResultCountLeavesItUnchanged() throws
InterruptedException {
+ String marker = "agg-result-key-" + UUID.randomUUID();
+ saveProfileWithLastName(marker, "_all");
+ saveProfileWithLastName(marker, "Smith");
+ refreshPersistence(Profile.class);
+
+ Map<String, Long> byLastName = keepTrying("Terms aggregate should
return the regular last-name bucket",
+ () ->
persistenceService.aggregateQuery(firstNameEquals(marker), new
TermsAggregate("properties.lastName"),
+ Profile.ITEM_TYPE),
+ r -> r != null && Long.valueOf(1L).equals(r.get("Smith")),
+ DEFAULT_TRYING_TIMEOUT, DEFAULT_TRYING_TRIES);
+ Assert.assertEquals("_all must still be the current tenant's profile
count",
+
Long.valueOf(persistenceService.getAllItemsCount(Profile.ITEM_TYPE)),
byLastName.get("_all"));
+ Assert.assertEquals(Long.valueOf(2L), byLastName.get("_filtered"));
+ }
+
private Condition matchAllCondition() {
return new
Condition(definitionsService.getConditionType("matchAllCondition"));
}
+ private Condition firstNameEquals(String firstName) {
+ Condition condition = new
Condition(definitionsService.getConditionType("profilePropertyCondition"));
+ condition.setParameter("propertyName", "properties.firstName");
+ condition.setParameter("comparisonOperator", "equals");
+ condition.setParameter("propertyValue", firstName);
+ return condition;
+ }
+
+ private void saveProfileWithLastName(String firstName, String lastName) {
+ String itemId = "agg-result-key-" + UUID.randomUUID();
+ Profile profile = new Profile(itemId);
+ profile.setProperty("firstName", firstName);
+ profile.setProperty("lastName", lastName);
+ persistenceService.save(profile);
+ profileIds.add(itemId);
+ }
+
+ private void saveProfileWithVisitAndAge(String firstName, String
lastVisit, int age) {
+ String itemId = "agg-buckets-" + UUID.randomUUID();
+ Profile profile = new Profile(itemId);
+ profile.setProperty("firstName", firstName);
+ profile.setProperty("lastVisit", Date.from(Instant.parse(lastVisit)));
+ profile.setProperty("age", age);
+ persistenceService.save(profile);
+ profileIds.add(itemId);
+ }
+
+ private static NumericRange numericRange(String key, Double from, Double
to) {
+ NumericRange range = new NumericRange();
+ range.setKey(key);
+ range.setFrom(from);
+ range.setTo(to);
+ return range;
+ }
+
+ private static DateRange dateRange(String key, String from, String to) {
+ DateRange range = new DateRange();
+ range.setKey(key);
+ range.setFrom(from);
+ range.setTo(to);
+ return range;
+ }
+
private void saveProfileWithFirstName(String itemId, String firstName) {
Profile profile = new Profile(itemId);
profile.setProperty("firstName", firstName);
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 f5980a7ac..b27807fec 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
@@ -102,6 +102,10 @@ import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import static org.apache.unomi.api.tenants.TenantService.SYSTEM_TENANT;
+import static
org.apache.unomi.persistence.spi.aggregate.AggregateBuckets.addBucket;
+import static
org.apache.unomi.persistence.spi.aggregate.AggregateBuckets.bucketKey;
+import static
org.apache.unomi.persistence.spi.aggregate.AggregateBuckets.ipRangeBucketKey;
+import static
org.apache.unomi.persistence.spi.aggregate.AggregateBuckets.rangeBucketKey;
@SuppressWarnings("rawtypes")
public class ElasticSearchPersistenceServiceImpl implements
PersistenceService, SynchronousBundleListener, ManagedService {
@@ -2468,7 +2472,6 @@ public class ElasticSearchPersistenceServiceImpl
implements PersistenceService,
List<IpRangeAggregationRange> ranges = new
ArrayList<>();
for (IpRange range : ipRangeAggregate.getRanges()) {
if (range != null) {
- IpRangeAggregationRange.of(builder ->
builder.from(range.getFrom()).to(range.getTo()));
ranges.add(IpRangeAggregationRange.of(builder
-> builder.from(range.getFrom()).to(range.getTo())));
}
}
@@ -2572,47 +2575,19 @@ public class ElasticSearchPersistenceServiceImpl
implements PersistenceService,
aggregations = filterAggregate.aggregations();
}
}
- if (aggregations.get("buckets") != null) {
- if (aggQueryThrowOnMissingDocs) {
- Aggregate agg = aggregations.get("buckets");
- if (agg.isSterms()) {
- StringTermsAggregate terms =
aggregations.get("buckets").sterms();
- if (terms.docCountErrorUpperBound() > 0 ||
terms.sumOtherDocCount() > 0) {
- throw new
UnsupportedOperationException("Some docs are missing in aggregation query.
docCountError is:"
- + terms.docCountErrorUpperBound()
+ " sumOfOtherDocCounts:" + terms.sumOtherDocCount());
- }
- }
- // TODO check if needed for dTerms and lTerms
- }
-
- long totalDocCount = 0;
- Aggregate bucketsAggregate =
aggregations.get("buckets");
- if (bucketsAggregate.isMultiTerms()) {
- MultiTermsAggregate terms =
aggregations.get("buckets").multiTerms();
- for (MultiTermsBucket bucket :
terms.buckets().array()) {
- results.put(bucket.keyAsString(),
bucket.docCount());
- totalDocCount += bucket.docCount();
- }
- } else if (bucketsAggregate.isSterms()) {
+ Aggregate bucketsAggregate = aggregations.get("buckets");
+ if (bucketsAggregate != null) {
+ // TODO check if needed for dTerms and lTerms
+ if (aggQueryThrowOnMissingDocs &&
bucketsAggregate.isSterms()) {
StringTermsAggregate terms =
bucketsAggregate.sterms();
- for (StringTermsBucket bucket :
terms.buckets().array()) {
- results.put(bucket.key().stringValue(),
bucket.docCount());
- totalDocCount += bucket.docCount();
- }
- } else if (bucketsAggregate.isLterms()) {
- LongTermsAggregate terms =
bucketsAggregate.lterms();
- for (LongTermsBucket bucket :
terms.buckets().array()) {
- results.put(bucket.keyAsString(),
bucket.docCount());
- totalDocCount += bucket.docCount();
- }
- } else if (bucketsAggregate.isDateHistogram()){
- DateHistogramAggregate histogramAggregate =
bucketsAggregate.dateHistogram();
- for (DateHistogramBucket bucket :
histogramAggregate.buckets().array()) {
- results.put(bucket.keyAsString(),
bucket.docCount());
- totalDocCount += bucket.docCount();
+ if (terms.docCountErrorUpperBound() > 0 ||
terms.sumOtherDocCount() > 0) {
+ throw new UnsupportedOperationException("Some
docs are missing in aggregation query. docCountError is:"
+ + terms.docCountErrorUpperBound() + "
sumOfOtherDocCounts:" + terms.sumOtherDocCount());
}
}
+ long totalDocCount = addBucketCounts(bucketsAggregate,
aggregate, results);
+
Aggregate missingAgg = aggregations.get("missing");
if (missingAgg != null && missingAgg.isMissing()) {
MissingAggregate missing = missingAgg.missing();
@@ -2632,6 +2607,58 @@ public class ElasticSearchPersistenceServiceImpl
implements PersistenceService,
}.catchingExecuteInClassLoader(true);
}
+ /**
+ * Adds the buckets of a terms, date histogram, or numeric, date or IP
range aggregate to {@code results} and
+ * returns their total document count. {@code aggregate} supplies the keys
of IP ranges.
+ */
+ static long addBucketCounts(Aggregate bucketsAggregate, BaseAggregate
aggregate, Map<String, Long> results) {
+ long totalDocCount = 0;
+ if (bucketsAggregate.isMultiTerms()) {
+ for (MultiTermsBucket bucket :
bucketsAggregate.multiTerms().buckets().array()) {
+ totalDocCount += addBucket(results, bucket.keyAsString(),
bucket.docCount());
+ }
+ } else if (bucketsAggregate.isSterms()) {
+ for (StringTermsBucket bucket :
bucketsAggregate.sterms().buckets().array()) {
+ totalDocCount += addBucket(results,
bucket.key().stringValue(), bucket.docCount());
+ }
+ } else if (bucketsAggregate.isLterms()) {
+ for (LongTermsBucket bucket :
bucketsAggregate.lterms().buckets().array()) {
+ totalDocCount += addBucket(results,
bucketKey(bucket.keyAsString(), bucket.key()), bucket.docCount());
+ }
+ } else if (bucketsAggregate.isDterms()) {
+ for (DoubleTermsBucket bucket :
bucketsAggregate.dterms().buckets().array()) {
+ totalDocCount += addBucket(results,
bucketKey(bucket.keyAsString(), bucket.key()), bucket.docCount());
+ }
+ } else if (bucketsAggregate.isDateHistogram()) {
+ for (DateHistogramBucket bucket :
bucketsAggregate.dateHistogram().buckets().array()) {
+ totalDocCount += addBucket(results,
bucketKey(bucket.keyAsString(), bucket.key()), bucket.docCount());
+ }
+ } else if (bucketsAggregate.isRange()) {
+ totalDocCount +=
addRangeBucketCounts(bucketsAggregate.range().buckets().array(), results);
+ } else if (bucketsAggregate.isDateRange()) {
+ totalDocCount +=
addRangeBucketCounts(bucketsAggregate.dateRange().buckets().array(), results);
+ } else if (bucketsAggregate.isIpRange()) {
+ for (IpRangeBucket bucket :
bucketsAggregate.ipRange().buckets().array()) {
+ totalDocCount += addBucket(results,
ipRangeBucketKey(aggregate, bucket.key(), bucket.from(), bucket.to()),
+ bucket.docCount());
+ }
+ } else {
+ LOGGER.warn("Aggregate results of type {} on field {} are not
read, their buckets are left out",
+ bucketsAggregate._kind(), aggregate.getField());
+ }
+ return totalDocCount;
+ }
+
+ private static long addRangeBucketCounts(List<RangeBucket> buckets,
Map<String, Long> results) {
+ long totalDocCount = 0;
+ for (RangeBucket bucket : buckets) {
+ totalDocCount += addBucket(results,
+ rangeBucketKey(bucket.key(), bucket.fromAsString(),
bucket.from(), bucket.toAsString(), bucket.to()),
+ bucket.docCount());
+ }
+ return totalDocCount;
+ }
+
private <T extends Item> String[] getRouting(String fieldName, String[]
fieldValues, Class<T> clazz) {
String itemType = Item.getItemType(clazz);
String[] routing = null;
diff --git
a/persistence-elasticsearch/core/src/test/java/org/apache/unomi/persistence/elasticsearch/ElasticSearchAggregateBucketsTest.java
b/persistence-elasticsearch/core/src/test/java/org/apache/unomi/persistence/elasticsearch/ElasticSearchAggregateBucketsTest.java
new file mode 100644
index 000000000..bbf919169
--- /dev/null
+++
b/persistence-elasticsearch/core/src/test/java/org/apache/unomi/persistence/elasticsearch/ElasticSearchAggregateBucketsTest.java
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.unomi.persistence.elasticsearch;
+
+import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
+import co.elastic.clients.elasticsearch.core.SearchResponse;
+import co.elastic.clients.json.JsonpDeserializer;
+import co.elastic.clients.json.JsonpMapper;
+import co.elastic.clients.json.jackson.JacksonJsonpMapper;
+import org.apache.unomi.api.query.IpRange;
+import org.apache.unomi.persistence.spi.aggregate.BaseAggregate;
+import org.apache.unomi.persistence.spi.aggregate.IpRangeAggregate;
+import org.apache.unomi.persistence.spi.aggregate.TermsAggregate;
+import org.junit.Test;
+
+import java.io.StringReader;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Reads aggregate buckets from Elasticsearch response JSON, parsed by the
client as it would be for a real search.
+ */
+public class ElasticSearchAggregateBucketsTest {
+
+ private static final BaseAggregate TERMS = new
TermsAggregate("properties.field");
+
+ @Test
+ public void readsStringTerms() {
+ Map<String, Long> results = read("sterms",
"{\"doc_count_error_upper_bound\":0,\"sum_other_doc_count\":0,"
+ +
"\"buckets\":[{\"key\":\"Alice\",\"doc_count\":2},{\"key\":\"Bob\",\"doc_count\":1}]}",
TERMS, 3);
+ assertEquals(Long.valueOf(2L), results.get("Alice"));
+ assertEquals(Long.valueOf(1L), results.get("Bob"));
+ }
+
+ @Test
+ public void readsLongTermsWithoutKeyAsString() {
+ Map<String, Long> results = read("lterms",
"{\"doc_count_error_upper_bound\":0,\"sum_other_doc_count\":0,"
+ +
"\"buckets\":[{\"key\":30,\"doc_count\":2},{\"key\":70,\"doc_count\":1}]}",
TERMS, 3);
+ assertEquals(Long.valueOf(2L), results.get("30"));
+ assertEquals(Long.valueOf(1L), results.get("70"));
+ assertEquals(2, results.size());
+ }
+
+ @Test
+ public void readsDoubleTerms() {
+ Map<String, Long> results = read("dterms",
"{\"doc_count_error_upper_bound\":0,\"sum_other_doc_count\":0,"
+ + "\"buckets\":[{\"key\":2.5,\"doc_count\":4}]}", TERMS, 4);
+ assertEquals(Long.valueOf(4L), results.get("2.5"));
+ }
+
+ @Test
+ public void readsDateHistogramWithFormattedKeys() {
+ Map<String, Long> results = read("date_histogram", "{\"buckets\":["
+ +
"{\"key_as_string\":\"2020-01-01\",\"key\":1577836800000,\"doc_count\":3},"
+ +
"{\"key_as_string\":\"2020-01-02\",\"key\":1577923200000,\"doc_count\":0}]}",
TERMS, 3);
+ assertEquals(Long.valueOf(3L), results.get("2020-01-01"));
+ assertEquals(Long.valueOf(0L), results.get("2020-01-02"));
+ }
+
+ @Test
+ public void readsNumericRanges() {
+ Map<String, Long> results = read("range", "{\"buckets\":["
+ + "{\"key\":\"young\",\"to\":18.0,\"doc_count\":1},"
+ +
"{\"key\":\"adult\",\"from\":18.0,\"to\":65.0,\"doc_count\":2},"
+ + "{\"from\":65.0,\"doc_count\":1}]}", TERMS, 4);
+ assertEquals(Long.valueOf(1L), results.get("young"));
+ assertEquals(Long.valueOf(2L), results.get("adult"));
+ assertEquals(Long.valueOf(1L), results.get("65.0-*"));
+ }
+
+ @Test
+ public void readsDateRanges() {
+ Map<String, Long> results = read("date_range", "{\"buckets\":["
+ +
"{\"key\":\"before\",\"to\":1577923200000,\"to_as_string\":\"2020-01-02\",\"doc_count\":3},"
+ +
"{\"from\":1577923200000,\"from_as_string\":\"2020-01-02\",\"doc_count\":1}]}",
TERMS, 4);
+ assertEquals(Long.valueOf(3L), results.get("before"));
+ assertEquals(Long.valueOf(1L), results.get("2020-01-02-*"));
+ }
+
+ @Test
+ public void readsIpRangesUnderTheirRangeKeys() {
+ IpRangeAggregate aggregate = new IpRangeAggregate("properties.ip",
+ Arrays.asList(ipRange("lan", "10.0.0.0", "10.0.1.0"),
ipRange("loopback", "0:0:0:0:0:0:0:1", null)));
+ Map<String, Long> results = read("ip_range", "{\"buckets\":["
+ +
"{\"key\":\"10.0.0.0-10.0.1.0\",\"from\":\"10.0.0.0\",\"to\":\"10.0.1.0\",\"doc_count\":2},"
+ + "{\"key\":\"::1-*\",\"from\":\"::1\",\"doc_count\":1}]}",
aggregate, 3);
+ assertEquals(Long.valueOf(2L), results.get("lan"));
+ assertEquals(Long.valueOf(1L), results.get("loopback"));
+ assertEquals(2, results.size());
+ }
+
+ @Test
+ public void readsMultiTerms() {
+ Map<String, Long> results = read("multi_terms",
"{\"doc_count_error_upper_bound\":0,\"sum_other_doc_count\":0,"
+ +
"\"buckets\":[{\"key\":[\"a\",\"b\"],\"key_as_string\":\"a|b\",\"doc_count\":1}]}",
TERMS, 1);
+ assertEquals(Long.valueOf(1L), results.get("a|b"));
+ }
+
+ @Test
+ public void leavesOutUnreadBucketTypes() {
+ Map<String, Long> results = read("avg", "{\"value\":1.5}", TERMS, 0);
+ assertTrue(results.isEmpty());
+ }
+
+ @Test
+ public void bucketsNamedLikeResultCountsDoNotChangeThem() {
+ Map<String, Long> results = new LinkedHashMap<>();
+ results.put("_all", 10L);
+ long total =
ElasticSearchPersistenceServiceImpl.addBucketCounts(parse("sterms",
+ "{\"doc_count_error_upper_bound\":0,\"sum_other_doc_count\":0,"
+ +
"\"buckets\":[{\"key\":\"_all\",\"doc_count\":3},{\"key\":\"x\",\"doc_count\":1}]}"),
TERMS, results);
+ assertEquals(4L, total);
+ assertEquals(Long.valueOf(10L), results.get("_all"));
+ assertEquals(Long.valueOf(1L), results.get("x"));
+ }
+
+ private static Map<String, Long> read(String type, String body,
BaseAggregate aggregate, long expectedTotal) {
+ Map<String, Long> results = new LinkedHashMap<>();
+ assertEquals(expectedTotal,
ElasticSearchPersistenceServiceImpl.addBucketCounts(parse(type, body),
aggregate, results));
+ return results;
+ }
+
+ private static Aggregate parse(String type, String body) {
+ String json = "{\"took\":1,\"timed_out\":false,"
+ +
"\"_shards\":{\"total\":1,\"successful\":1,\"skipped\":0,\"failed\":0},"
+ +
"\"hits\":{\"total\":{\"value\":0,\"relation\":\"eq\"},\"hits\":[]},"
+ + "\"aggregations\":{\"" + type + "#buckets\":" + body + "}}";
+ JsonpMapper mapper = new JacksonJsonpMapper();
+ SearchResponse<Void> response =
SearchResponse.createSearchResponseDeserializer(JsonpDeserializer.voidDeserializer())
+ .deserialize(mapper.jsonProvider().createParser(new
StringReader(json)), mapper);
+ return response.aggregations().get("buckets");
+ }
+
+ private static IpRange ipRange(String key, String from, String to) {
+ IpRange range = new IpRange();
+ range.setKey(key);
+ range.setFrom(from);
+ range.setTo(to);
+ return range;
+ }
+}
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 0e2be7a7e..235818bd2 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
@@ -98,6 +98,10 @@ import java.util.concurrent.CopyOnWriteArrayList;
import java.util.stream.Collectors;
import static org.apache.unomi.api.tenants.TenantService.SYSTEM_TENANT;
+import static
org.apache.unomi.persistence.spi.aggregate.AggregateBuckets.addBucket;
+import static
org.apache.unomi.persistence.spi.aggregate.AggregateBuckets.bucketKey;
+import static
org.apache.unomi.persistence.spi.aggregate.AggregateBuckets.ipRangeBucketKey;
+import static
org.apache.unomi.persistence.spi.aggregate.AggregateBuckets.rangeBucketKey;
@SuppressWarnings("rawtypes")
public class OpenSearchPersistenceServiceImpl implements PersistenceService,
SynchronousBundleListener, ManagedService {
@@ -2416,7 +2420,7 @@ public class OpenSearchPersistenceServiceImpl implements
PersistenceService, Syn
DateAggregate dateAggregate = (DateAggregate)
aggregate;
DateHistogramAggregation.Builder dateHistogramBuilder
= new DateHistogramAggregation.Builder()
.field(fieldName)
-
.calendarInterval(CalendarInterval.valueOf(dateAggregate.getInterval()));
+
.calendarInterval(CalendarInterval.valueOf(dateAggregate.getIntervalByAlias(dateAggregate.getInterval())));
if (dateAggregate.getFormat() != null) {
dateHistogramBuilder.format(dateAggregate.getFormat());
}
@@ -2426,17 +2430,16 @@ public class OpenSearchPersistenceServiceImpl
implements PersistenceService, Syn
.field(fieldName);
for (NumericRange range : ((NumericRangeAggregate)
aggregate).getRanges()) {
if (range != null) {
- if (range.getFrom() != null && range.getTo()
!= null) {
- rangeBuilder.ranges(r ->
r.key(range.getKey())
- .from(JsonData.of(range.getFrom()))
- .to(JsonData.of(range.getTo())));
- } else if (range.getFrom() != null) {
- rangeBuilder.ranges(r ->
r.key(range.getKey())
-
.from(JsonData.of(range.getFrom())));
- } else if (range.getTo() != null) {
- rangeBuilder.ranges(r ->
r.key(range.getKey())
- .to(JsonData.of(range.getTo())));
- }
+ rangeBuilder.ranges(r -> {
+ r.key(range.getKey());
+ if (range.getFrom() != null) {
+ r.from(JsonData.of(range.getFrom()));
+ }
+ if (range.getTo() != null) {
+ r.to(JsonData.of(range.getTo()));
+ }
+ return r;
+ });
}
}
bucketsContainerBuilder =
bucketsAggregationBuilder.range(rangeBuilder.build());
@@ -2449,9 +2452,16 @@ public class OpenSearchPersistenceServiceImpl implements
PersistenceService, Syn
}
for (DateRange range :
dateRangeAggregate.getDateRanges()) {
if (range != null) {
- rangeBuilder.ranges(r -> r.key(range.getKey())
- .from(f -> f.expr(range.getFrom() !=
null ? range.getFrom().toString() : null))
- .to(t -> t.expr(range.getTo() != null
? range.getTo().toString() : null)));
+ rangeBuilder.ranges(r -> {
+ r.key(range.getKey());
+ if (range.getFrom() != null) {
+ r.from(f ->
f.expr(range.getFrom().toString()));
+ }
+ if (range.getTo() != null) {
+ r.to(t ->
t.expr(range.getTo().toString()));
+ }
+ return r;
+ });
}
}
bucketsContainerBuilder =
bucketsAggregationBuilder.dateRange(rangeBuilder.build());
@@ -2461,7 +2471,7 @@ public class OpenSearchPersistenceServiceImpl implements
PersistenceService, Syn
.field(fieldName);
for (IpRange range : ipRangeAggregate.getRanges()) {
if (range != null) {
- rangeBuilder.ranges(r ->
r.mask(range.getKey()).from(range.getFrom()).to(range.getTo()));
+ rangeBuilder.ranges(r ->
r.from(range.getFrom()).to(range.getTo()));
}
}
bucketsContainerBuilder =
bucketsAggregationBuilder.ipRange(rangeBuilder.build());
@@ -2565,26 +2575,17 @@ public class OpenSearchPersistenceServiceImpl
implements PersistenceService, Syn
aggregates = filterAgg.aggregations();
}
}
- if (aggregates.get("buckets") != null) {
-
- if (aggQueryThrowOnMissingDocs) {
- if (aggregates.get("buckets").isSterms()) {
- StringTermsAggregate terms =
aggregates.get("buckets").sterms();
- if (terms.docCountErrorUpperBound() > 0 ||
terms.sumOtherDocCount() > 0) {
- throw new
UnsupportedOperationException("Some docs are missing in aggregation query.
docCountError is:" +
- terms.docCountErrorUpperBound() +
" sumOfOtherDocCounts:" + terms.sumOtherDocCount());
- }
+ Aggregate bucketsAggregate = aggregates.get("buckets");
+ if (bucketsAggregate != null) {
+ if (aggQueryThrowOnMissingDocs &&
bucketsAggregate.isSterms()) {
+ StringTermsAggregate terms =
bucketsAggregate.sterms();
+ if (terms.docCountErrorUpperBound() > 0 ||
terms.sumOtherDocCount() > 0) {
+ throw new UnsupportedOperationException("Some
docs are missing in aggregation query. docCountError is:" +
+ terms.docCountErrorUpperBound() + "
sumOfOtherDocCounts:" + terms.sumOtherDocCount());
}
}
- long totalDocCount = 0;
- Aggregate bucketsAggregate = aggregates.get("buckets");
- if (bucketsAggregate.isSterms()) {
- for (StringTermsBucket bucket :
bucketsAggregate.sterms().buckets().array()) {
- results.put(bucket.key(), bucket.docCount());
- totalDocCount += bucket.docCount();
- }
- }
+ long totalDocCount = addBucketCounts(bucketsAggregate,
aggregate, results);
Aggregate missingAgg = aggregates.get("missing");
if (missingAgg != null && missingAgg.isMissing()) {
MissingAggregate missing = missingAgg.missing();
@@ -2604,6 +2605,54 @@ public class OpenSearchPersistenceServiceImpl implements
PersistenceService, Syn
}.catchingExecuteInClassLoader(true);
}
+ /**
+ * Adds the buckets of a terms, date histogram, or numeric, date or IP
range aggregate to {@code results} and
+ * returns their total document count. {@code aggregate} supplies the keys
of IP ranges.
+ */
+ static long addBucketCounts(Aggregate bucketsAggregate, BaseAggregate
aggregate, Map<String, Long> results) {
+ long totalDocCount = 0;
+ if (bucketsAggregate.isSterms()) {
+ for (StringTermsBucket bucket :
bucketsAggregate.sterms().buckets().array()) {
+ totalDocCount += addBucket(results, bucket.key(),
bucket.docCount());
+ }
+ } else if (bucketsAggregate.isLterms()) {
+ for (LongTermsBucket bucket :
bucketsAggregate.lterms().buckets().array()) {
+ totalDocCount += addBucket(results,
bucketKey(bucket.keyAsString(), bucket.key()._get()), bucket.docCount());
+ }
+ } else if (bucketsAggregate.isDterms()) {
+ for (DoubleTermsBucket bucket :
bucketsAggregate.dterms().buckets().array()) {
+ totalDocCount += addBucket(results,
bucketKey(bucket.keyAsString(), bucket.key()), bucket.docCount());
+ }
+ } else if (bucketsAggregate.isDateHistogram()) {
+ for (DateHistogramBucket bucket :
bucketsAggregate.dateHistogram().buckets().array()) {
+ totalDocCount += addBucket(results,
bucketKey(bucket.keyAsString(), bucket.key()), bucket.docCount());
+ }
+ } else if (bucketsAggregate.isRange()) {
+ totalDocCount +=
addRangeBucketCounts(bucketsAggregate.range().buckets().array(), results);
+ } else if (bucketsAggregate.isDateRange()) {
+ totalDocCount +=
addRangeBucketCounts(bucketsAggregate.dateRange().buckets().array(), results);
+ } else if (bucketsAggregate.isIpRange()) {
+ for (IpRangeBucket bucket :
bucketsAggregate.ipRange().buckets().array()) {
+ totalDocCount += addBucket(results,
ipRangeBucketKey(aggregate, bucket.key(), bucket.from(), bucket.to()),
+ bucket.docCount());
+ }
+ } else {
+ LOGGER.warn("Aggregate results of type {} on field {} are not
read, their buckets are left out",
+ bucketsAggregate._kind(), aggregate.getField());
+ }
+ return totalDocCount;
+ }
+
+ private static long addRangeBucketCounts(List<RangeBucket> buckets,
Map<String, Long> results) {
+ long totalDocCount = 0;
+ for (RangeBucket bucket : buckets) {
+ totalDocCount += addBucket(results,
+ rangeBucketKey(bucket.key(), bucket.fromAsString(),
bucket.from(), bucket.toAsString(), bucket.to()),
+ bucket.docCount());
+ }
+ return totalDocCount;
+ }
+
private <T extends Item> String[] getRouting(String fieldName, String[]
fieldValues, Class<T> clazz) {
String itemType = Item.getItemType(clazz);
String[] routing = null;
diff --git
a/persistence-opensearch/core/src/test/java/org/apache/unomi/persistence/opensearch/OpenSearchAggregateBucketsTest.java
b/persistence-opensearch/core/src/test/java/org/apache/unomi/persistence/opensearch/OpenSearchAggregateBucketsTest.java
new file mode 100644
index 000000000..52c412426
--- /dev/null
+++
b/persistence-opensearch/core/src/test/java/org/apache/unomi/persistence/opensearch/OpenSearchAggregateBucketsTest.java
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.unomi.persistence.opensearch;
+
+import org.opensearch.client.opensearch._types.aggregations.Aggregate;
+import org.opensearch.client.opensearch.core.SearchResponse;
+import org.opensearch.client.json.JsonpDeserializer;
+import org.opensearch.client.json.JsonpMapper;
+import org.opensearch.client.json.jackson.JacksonJsonpMapper;
+import org.apache.unomi.api.query.IpRange;
+import org.apache.unomi.persistence.spi.aggregate.BaseAggregate;
+import org.apache.unomi.persistence.spi.aggregate.IpRangeAggregate;
+import org.apache.unomi.persistence.spi.aggregate.TermsAggregate;
+import org.junit.jupiter.api.Test;
+
+import java.io.StringReader;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Reads aggregate buckets from OpenSearch response JSON, parsed by the client
as it would be for a real search.
+ */
+public class OpenSearchAggregateBucketsTest {
+
+ private static final BaseAggregate TERMS = new
TermsAggregate("properties.field");
+
+ @Test
+ public void readsStringTerms() {
+ Map<String, Long> results = read("sterms",
"{\"doc_count_error_upper_bound\":0,\"sum_other_doc_count\":0,"
+ +
"\"buckets\":[{\"key\":\"Alice\",\"doc_count\":2},{\"key\":\"Bob\",\"doc_count\":1}]}",
TERMS, 3);
+ assertEquals(Long.valueOf(2L), results.get("Alice"));
+ assertEquals(Long.valueOf(1L), results.get("Bob"));
+ }
+
+ @Test
+ public void readsLongTermsWithoutKeyAsString() {
+ Map<String, Long> results = read("lterms",
"{\"doc_count_error_upper_bound\":0,\"sum_other_doc_count\":0,"
+ +
"\"buckets\":[{\"key\":30,\"doc_count\":2},{\"key\":70,\"doc_count\":1}]}",
TERMS, 3);
+ assertEquals(Long.valueOf(2L), results.get("30"));
+ assertEquals(Long.valueOf(1L), results.get("70"));
+ assertEquals(2, results.size());
+ }
+
+ @Test
+ public void readsDoubleTerms() {
+ Map<String, Long> results = read("dterms",
"{\"doc_count_error_upper_bound\":0,\"sum_other_doc_count\":0,"
+ + "\"buckets\":[{\"key\":2.5,\"doc_count\":4}]}", TERMS, 4);
+ assertEquals(Long.valueOf(4L), results.get("2.5"));
+ }
+
+ @Test
+ public void readsDateHistogramWithFormattedKeys() {
+ Map<String, Long> results = read("date_histogram", "{\"buckets\":["
+ +
"{\"key_as_string\":\"2020-01-01\",\"key\":1577836800000,\"doc_count\":3},"
+ +
"{\"key_as_string\":\"2020-01-02\",\"key\":1577923200000,\"doc_count\":0}]}",
TERMS, 3);
+ assertEquals(Long.valueOf(3L), results.get("2020-01-01"));
+ assertEquals(Long.valueOf(0L), results.get("2020-01-02"));
+ }
+
+ @Test
+ public void readsNumericRanges() {
+ Map<String, Long> results = read("range", "{\"buckets\":["
+ + "{\"key\":\"young\",\"to\":18.0,\"doc_count\":1},"
+ +
"{\"key\":\"adult\",\"from\":18.0,\"to\":65.0,\"doc_count\":2},"
+ + "{\"from\":65.0,\"doc_count\":1}]}", TERMS, 4);
+ assertEquals(Long.valueOf(1L), results.get("young"));
+ assertEquals(Long.valueOf(2L), results.get("adult"));
+ assertEquals(Long.valueOf(1L), results.get("65.0-*"));
+ }
+
+ @Test
+ public void readsDateRanges() {
+ Map<String, Long> results = read("date_range", "{\"buckets\":["
+ +
"{\"key\":\"before\",\"to\":1577923200000,\"to_as_string\":\"2020-01-02\",\"doc_count\":3},"
+ +
"{\"from\":1577923200000,\"from_as_string\":\"2020-01-02\",\"doc_count\":1}]}",
TERMS, 4);
+ assertEquals(Long.valueOf(3L), results.get("before"));
+ assertEquals(Long.valueOf(1L), results.get("2020-01-02-*"));
+ }
+
+ @Test
+ public void readsIpRangesUnderTheirRangeKeys() {
+ IpRangeAggregate aggregate = new IpRangeAggregate("properties.ip",
+ Arrays.asList(ipRange("lan", "10.0.0.0", "10.0.1.0"),
ipRange("loopback", "0:0:0:0:0:0:0:1", null)));
+ Map<String, Long> results = read("ip_range", "{\"buckets\":["
+ +
"{\"key\":\"10.0.0.0-10.0.1.0\",\"from\":\"10.0.0.0\",\"to\":\"10.0.1.0\",\"doc_count\":2},"
+ + "{\"key\":\"::1-*\",\"from\":\"::1\",\"doc_count\":1}]}",
aggregate, 3);
+ assertEquals(Long.valueOf(2L), results.get("lan"));
+ assertEquals(Long.valueOf(1L), results.get("loopback"));
+ assertEquals(2, results.size());
+ }
+
+ @Test
+ public void leavesOutUnreadBucketTypes() {
+ Map<String, Long> results = read("avg", "{\"value\":1.5}", TERMS, 0);
+ assertTrue(results.isEmpty());
+ }
+
+ @Test
+ public void bucketsNamedLikeResultCountsDoNotChangeThem() {
+ Map<String, Long> results = new LinkedHashMap<>();
+ results.put("_all", 10L);
+ long total =
OpenSearchPersistenceServiceImpl.addBucketCounts(parse("sterms",
+ "{\"doc_count_error_upper_bound\":0,\"sum_other_doc_count\":0,"
+ +
"\"buckets\":[{\"key\":\"_all\",\"doc_count\":3},{\"key\":\"x\",\"doc_count\":1}]}"),
TERMS, results);
+ assertEquals(4L, total);
+ assertEquals(Long.valueOf(10L), results.get("_all"));
+ assertEquals(Long.valueOf(1L), results.get("x"));
+ }
+
+ private static Map<String, Long> read(String type, String body,
BaseAggregate aggregate, long expectedTotal) {
+ Map<String, Long> results = new LinkedHashMap<>();
+ assertEquals(expectedTotal,
OpenSearchPersistenceServiceImpl.addBucketCounts(parse(type, body), aggregate,
results));
+ return results;
+ }
+
+ private static Aggregate parse(String type, String body) {
+ String json = "{\"took\":1,\"timed_out\":false,"
+ +
"\"_shards\":{\"total\":1,\"successful\":1,\"skipped\":0,\"failed\":0},"
+ +
"\"hits\":{\"total\":{\"value\":0,\"relation\":\"eq\"},\"hits\":[]},"
+ + "\"aggregations\":{\"" + type + "#buckets\":" + body + "}}";
+ JsonpMapper mapper = new JacksonJsonpMapper();
+ SearchResponse<Void> response =
SearchResponse.createSearchResponseDeserializer(JsonpDeserializer.voidDeserializer())
+ .deserialize(mapper.jsonProvider().createParser(new
StringReader(json)), mapper);
+ return response.aggregations().get("buckets");
+ }
+
+ private static IpRange ipRange(String key, String from, String to) {
+ IpRange range = new IpRange();
+ range.setKey(key);
+ range.setFrom(from);
+ range.setTo(to);
+ return range;
+ }
+}
diff --git
a/persistence-spi/src/main/java/org/apache/unomi/persistence/spi/aggregate/AggregateBuckets.java
b/persistence-spi/src/main/java/org/apache/unomi/persistence/spi/aggregate/AggregateBuckets.java
new file mode 100644
index 000000000..9f7329ea8
--- /dev/null
+++
b/persistence-spi/src/main/java/org/apache/unomi/persistence/spi/aggregate/AggregateBuckets.java
@@ -0,0 +1,171 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.unomi.persistence.spi.aggregate;
+
+import org.apache.unomi.api.query.IpRange;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+/**
+ * Turns the buckets a search engine returns for an aggregate into the keys
and counts of
+ * {@link org.apache.unomi.persistence.spi.PersistenceService} aggregate
results, so that every persistence
+ * implementation reports buckets the same way.
+ * <p>
+ * Support code for persistence implementations, not intended for other
callers.
+ */
+public final class AggregateBuckets {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(AggregateBuckets.class);
+
+ /**
+ * Keys that aggregate results use for their own counts rather than for
buckets.
+ */
+ private static final Set<String> RESULT_COUNT_KEYS = Set.of("_all",
"_filtered", "_missing");
+
+ private static final String OCTET =
"(25[0-5]|2[0-4]\\d|1\\d\\d|[1-9]?\\d)";
+ private static final Pattern IPV4_LITERAL = Pattern.compile(OCTET + "(\\."
+ OCTET + "){3}");
+ // Starts with a hex digit or ':' and contains ':'. The JDK parses such a
string as an IPv6 literal or rejects it,
+ // without a lookup. It only resolves names that start with another
character. Scoped addresses (with '%') do not
+ // match, so they are only compared as plain strings.
+ private static final Pattern IPV6_LITERAL =
Pattern.compile("(?=.*:)[0-9A-Fa-f:][0-9A-Fa-f:.]*");
+
+ private AggregateBuckets() {
+ }
+
+ /**
+ * Key of a terms or date histogram bucket. Engines send {@code
key_as_string} for formatted values such as
+ * dates, and omit it for plain numeric terms, in which case the raw key
is used.
+ *
+ * @param keyAsString the formatted key sent by the engine, or {@code null}
+ * @param key the raw key
+ * @return the bucket key
+ */
+ public static String bucketKey(String keyAsString, Object key) {
+ return keyAsString != null ? keyAsString : String.valueOf(key);
+ }
+
+ /**
+ * Key of a numeric or date range bucket: the engine's key, or else {@code
from-to} with {@code *} for an open
+ * bound, which is the form the engines generate for unnamed ranges.
+ *
+ * @param key the key sent by the engine, or {@code null}
+ * @param fromAsString the formatted lower bound, or {@code null}
+ * @param from the raw lower bound, or {@code null} if open
+ * @param toAsString the formatted upper bound, or {@code null}
+ * @param to the raw upper bound, or {@code null} if open
+ * @return the bucket key
+ */
+ public static String rangeBucketKey(String key, String fromAsString,
Object from, String toAsString, Object to) {
+ if (key != null) {
+ return key;
+ }
+ return rangeKey(fromAsString != null ? fromAsString :
Objects.toString(from, null),
+ toAsString != null ? toAsString : Objects.toString(to, null));
+ }
+
+ /**
+ * Key of an IP range bucket. IP range requests cannot carry a key, so the
bucket is reported under the key of the
+ * {@link IpRange} whose bounds match the bucket's. Any other bucket keeps
the engine's key, or {@code from-to} when
+ * the engine sends none. A bucket whose bounds match no range at all is
logged, since it cannot be labelled.
+ *
+ * @param aggregate the aggregate the request was built from
+ * @param key the key sent by the engine, or {@code null}
+ * @param from the lower bound sent by the engine, or {@code null} if
open
+ * @param to the upper bound sent by the engine, or {@code null} if
open
+ * @return the bucket key
+ */
+ public static String ipRangeBucketKey(BaseAggregate aggregate, String key,
String from, String to) {
+ boolean hasKeyedRange = false;
+ boolean matchesUnkeyedRange = false;
+ if (aggregate instanceof IpRangeAggregate && ((IpRangeAggregate)
aggregate).getRanges() != null) {
+ for (IpRange range : ((IpRangeAggregate) aggregate).getRanges()) {
+ if (range == null) {
+ continue;
+ }
+ boolean sameBounds = sameIp(range.getFrom(), from) &&
sameIp(range.getTo(), to);
+ if (range.getKey() != null) {
+ hasKeyedRange = true;
+ if (sameBounds) {
+ return range.getKey();
+ }
+ } else if (sameBounds) {
+ matchesUnkeyedRange = true;
+ }
+ }
+ }
+ String engineKey = key != null ? key : rangeKey(from, to);
+ if (hasKeyedRange && !matchesUnkeyedRange) {
+ LOGGER.warn("IP range bucket {} on field {} matches no keyed
range, reporting it under the engine key",
+ engineKey, aggregate.getField());
+ }
+ return engineKey;
+ }
+
+ /**
+ * Adds a bucket count to aggregate results. Buckets that share a key,
such as two ranges with the same key, are
+ * summed so that the results stay consistent with the total document
count. A bucket named like one of the
+ * result counts ({@code _all}, {@code _filtered}, {@code _missing}) is
left out and logged, so that it cannot
+ * change those counts.
+ *
+ * @param results the aggregate results
+ * @param key the bucket key
+ * @param docCount the bucket document count
+ * @return {@code docCount}, which still counts towards the total document
count
+ */
+ public static long addBucket(Map<String, Long> results, String key, long
docCount) {
+ if (key != null && RESULT_COUNT_KEYS.contains(key)) {
+ LOGGER.warn("Aggregate bucket {} has the name of a result count,
its {} documents are left out of the results",
+ key, docCount);
+ return docCount;
+ }
+ results.merge(key, docCount, Long::sum);
+ return docCount;
+ }
+
+ private static String rangeKey(String from, String to) {
+ return (from != null ? from : "*") + "-" + (to != null ? to : "*");
+ }
+
+ /**
+ * Engines return IPv6 bounds in canonical form (e.g. {@code ::1}), so
compare addresses rather than strings.
+ * Only strict IP literals are parsed: for those the JDK never falls back
to a DNS lookup.
+ */
+ static boolean sameIp(String a, String b) {
+ if (Objects.equals(a, b)) {
+ return true;
+ }
+ if (a == null || b == null || !isIpLiteral(a) || !isIpLiteral(b)) {
+ return false;
+ }
+ try {
+ return InetAddress.getByName(a).equals(InetAddress.getByName(b));
+ } catch (UnknownHostException e) {
+ return false;
+ }
+ }
+
+ static boolean isIpLiteral(String value) {
+ return IPV4_LITERAL.matcher(value).matches() ||
IPV6_LITERAL.matcher(value).matches();
+ }
+}
diff --git
a/persistence-spi/src/test/java/org/apache/unomi/persistence/spi/aggregate/AggregateBucketsTest.java
b/persistence-spi/src/test/java/org/apache/unomi/persistence/spi/aggregate/AggregateBucketsTest.java
new file mode 100644
index 000000000..fffefa6b9
--- /dev/null
+++
b/persistence-spi/src/test/java/org/apache/unomi/persistence/spi/aggregate/AggregateBucketsTest.java
@@ -0,0 +1,154 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.unomi.persistence.spi.aggregate;
+
+import org.apache.unomi.api.query.IpRange;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+public class AggregateBucketsTest {
+
+ @Test
+ public void bucketKeyPrefersTheFormattedKey() {
+ assertEquals("2020-01-01", AggregateBuckets.bucketKey("2020-01-01",
1577836800000L));
+ assertEquals("70", AggregateBuckets.bucketKey(null, 70L));
+ assertEquals("2.5", AggregateBuckets.bucketKey(null, 2.5d));
+ }
+
+ @Test
+ public void rangeBucketKeyFallsBackToTheEngineFormat() {
+ assertEquals("adult", AggregateBuckets.rangeBucketKey("adult", null,
18.0, null, 65.0));
+ assertEquals("18.0-65.0", AggregateBuckets.rangeBucketKey(null, null,
18.0, null, 65.0));
+ assertEquals("*-2020-01-02", AggregateBuckets.rangeBucketKey(null,
null, null, "2020-01-02", 1577923200000.0));
+ assertEquals("*-*", AggregateBuckets.rangeBucketKey(null, null, null,
null, null));
+ assertEquals("2020-01-01-*", AggregateBuckets.rangeBucketKey(null,
"2020-01-01", 1577836800000.0, null, null));
+ }
+
+ @Test
+ public void ipRangeBucketKeyUsesTheMatchingRangeKey() {
+ IpRangeAggregate aggregate = new IpRangeAggregate("properties.ip",
+ Arrays.asList(ipRange("lan", "10.0.0.0", "10.0.1.0"),
ipRange("loopback", "::1", null), null));
+
+ assertEquals("lan", AggregateBuckets.ipRangeBucketKey(aggregate,
"10.0.0.0-10.0.1.0", "10.0.0.0", "10.0.1.0"));
+ // engines return IPv6 bounds in canonical form
+ assertEquals("loopback", AggregateBuckets.ipRangeBucketKey(
+ new IpRangeAggregate("properties.ip",
Arrays.asList(ipRange("loopback", "0:0:0:0:0:0:0:1", null))),
+ "::1-*", "::1", null));
+ }
+
+ @Test
+ public void ipRangeBucketKeyKeepsTheEngineKeyWithoutAMatch() {
+ IpRangeAggregate aggregate = new IpRangeAggregate("properties.ip",
+ Arrays.asList(ipRange("lan", "10.0.0.0", "10.0.1.0")));
+
+ assertEquals("192.168.0.0-*",
AggregateBuckets.ipRangeBucketKey(aggregate, "192.168.0.0-*", "192.168.0.0",
null));
+ assertEquals("192.168.0.0-*",
AggregateBuckets.ipRangeBucketKey(aggregate, null, "192.168.0.0", null));
+ }
+
+ @Test
+ public void ipRangeBucketKeyKeepsTheEngineKeyOfAnUnkeyedRange() {
+ IpRangeAggregate aggregate = new IpRangeAggregate("properties.ip",
+ Arrays.asList(ipRange("lan", "10.0.0.0", "10.0.1.0"),
ipRange(null, "192.168.0.0", null)));
+
+ assertEquals("192.168.0.0-*",
AggregateBuckets.ipRangeBucketKey(aggregate, "192.168.0.0-*", "192.168.0.0",
null));
+ }
+
+ @Test
+ public void ipRangeBucketKeyKeepsTheEngineKeyWithoutIpRanges() {
+ assertEquals("10.0.0.0-10.0.1.0", AggregateBuckets.ipRangeBucketKey(
+ new TermsAggregate("properties.ip"), "10.0.0.0-10.0.1.0",
"10.0.0.0", "10.0.1.0"));
+ assertEquals("1.1.1.1-*", AggregateBuckets.ipRangeBucketKey(
+ new IpRangeAggregate("properties.ip", null), null, "1.1.1.1",
null));
+ assertEquals("1.1.1.1-*", AggregateBuckets.ipRangeBucketKey(null,
"1.1.1.1-*", "1.1.1.1", null));
+ }
+
+ @Test
+ public void addBucketSumsBucketsThatShareAKey() {
+ Map<String, Long> results = new LinkedHashMap<>();
+ assertEquals(2L, AggregateBuckets.addBucket(results, "adult", 2L));
+ assertEquals(3L, AggregateBuckets.addBucket(results, "adult", 3L));
+ assertEquals(Long.valueOf(5L), results.get("adult"));
+ }
+
+ @Test
+ public void addBucketLeavesResultCountsUntouched() {
+ Map<String, Long> results = new LinkedHashMap<>();
+ results.put("_all", 10L);
+ results.put("_filtered", 4L);
+
+ assertEquals(3L, AggregateBuckets.addBucket(results, "_all", 3L));
+ assertEquals(1L, AggregateBuckets.addBucket(results, "_filtered", 1L));
+ assertEquals(2L, AggregateBuckets.addBucket(results, "_missing", 2L));
+
+ assertEquals(Long.valueOf(10L), results.get("_all"));
+ assertEquals(Long.valueOf(4L), results.get("_filtered"));
+ assertNull(results.get("_missing"));
+ assertEquals(2, results.size());
+ }
+
+ @Test
+ public void addBucketAcceptsANullKey() {
+ Map<String, Long> results = new LinkedHashMap<>();
+ assertEquals(1L, AggregateBuckets.addBucket(results, null, 1L));
+ assertEquals(Long.valueOf(1L), results.get(null));
+ }
+
+ @Test
+ public void sameIpComparesAddresses() {
+ assertTrue(AggregateBuckets.sameIp("::1", "0:0:0:0:0:0:0:1"));
+ assertTrue(AggregateBuckets.sameIp("1.2.3.4", "::ffff:1.2.3.4"));
+ assertTrue(AggregateBuckets.sameIp(null, null));
+ assertFalse(AggregateBuckets.sameIp("10.0.0.1", "10.0.0.2"));
+ assertFalse(AggregateBuckets.sameIp("10.0.0.1", null));
+ assertFalse(AggregateBuckets.sameIp("localhost", "127.0.0.1"));
+ // scoped addresses are not parsed, only compared as strings
+ assertTrue(AggregateBuckets.sameIp("fe80::1%eth0", "fe80::1%eth0"));
+ assertFalse(AggregateBuckets.sameIp("fe80::1%eth0", "fe80::1"));
+ }
+
+ @Test
+ public void onlyStrictIpLiteralsAreParsed() {
+ assertTrue(AggregateBuckets.isIpLiteral("10.0.0.1"));
+ assertTrue(AggregateBuckets.isIpLiteral("::1"));
+ assertTrue(AggregateBuckets.isIpLiteral("fe80::1"));
+ // these would make InetAddress.getByName resolve a host name
+ assertFalse(AggregateBuckets.isIpLiteral("999.1.1.1"));
+ assertFalse(AggregateBuckets.isIpLiteral("localhost"));
+ assertFalse(AggregateBuckets.isIpLiteral("zz:1"));
+ assertFalse(AggregateBuckets.isIpLiteral(".:1"));
+ assertFalse(AggregateBuckets.isIpLiteral("fe80::1%eth0"));
+ // malformed IPv6 is rejected by the JDK without a lookup
+ assertTrue(AggregateBuckets.isIpLiteral("1::2::3"));
+ assertFalse(AggregateBuckets.sameIp("1::2::3", "::1"));
+ }
+
+ private static IpRange ipRange(String key, String from, String to) {
+ IpRange range = new IpRange();
+ range.setKey(key);
+ range.setFrom(from);
+ range.setTo(to);
+ return range;
+ }
+}
