UNOMI-948: Add PersistenceService.rangeQuery API

itests/src/test/java/org/apache/unomi/itests/AllITs.java

@@ -33,6 +33,7 @@
 @SuiteClasses({
         Migrate16xToCurrentVersionIT.class,
         MigrationIT.class,
+        PersistenceServiceIT.class,
         BasicIT.class,
         ConditionEvaluatorIT.class,
         ConditionQueryBuilderIT.class,

itests/src/test/java/org/apache/unomi/itests/PersistenceServiceIT.java

@@ -0,0 +1,115 @@
+/*
+ * 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.itests;
+
+import org.apache.unomi.api.PartialList;
+import org.apache.unomi.api.Profile;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+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.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * Integration tests for {@link org.apache.unomi.persistence.spi.PersistenceService} query APIs against the live
+ * search backend (Elasticsearch or OpenSearch). Initial coverage focuses on {@code rangeQuery}; additional methods
+ * should be covered in follow-up work (UNOMI-956).
+ */
+@RunWith(PaxExam.class)
+@ExamReactorStrategy(PerSuite.class)
+public class PersistenceServiceIT extends BaseIT {
+
+    private static final String AGE_PROPERTY = "properties.age";
+
+    private final List<String> profileIds = new ArrayList<>();
+
+    @After
+    public void tearDown() throws InterruptedException {
+        for (String profileId : profileIds) {
+            persistenceService.remove(profileId, Profile.class);
+        }
+        profileIds.clear();
+        refreshPersistence(Profile.class);
+    }
+
+    @Test
+    public void testRangeQueryReturnsProfilesInNumericRange() throws InterruptedException {
+        saveProfileWithAge("range-query-it-low", 10);
+        saveProfileWithAge("range-query-it-match-a", 20);
+        saveProfileWithAge("range-query-it-match-b", 30);
+        saveProfileWithAge("range-query-it-match-c", 40);
+        saveProfileWithAge("range-query-it-high", 50);
+
+        refreshPersistence(Profile.class);
+
+        // Both bounds are inclusive, so age 40 (the "to" value) must be included while age 50 is excluded.
+        PartialList<Profile> results = keepTrying(
+                "Range query should return profiles with age between 20 and 40",
+                () -> persistenceService.rangeQuery(AGE_PROPERTY, "20", "40", AGE_PROPERTY + ":asc", Profile.class, 0, -1),
+                r -> r != null && r.getList().size() == 3,
+                DEFAULT_TRYING_TIMEOUT, DEFAULT_TRYING_TRIES);
+
+        Assert.assertEquals(3, results.getList().size());
+        Assert.assertEquals(20, results.getList().get(0).getProperty("age"));
+        Assert.assertEquals(30, results.getList().get(1).getProperty("age"));
+        Assert.assertEquals(40, results.getList().get(2).getProperty("age"));
+    }
+
+    @Test
+    public void testRangeQuerySupportsPagination() throws InterruptedException {
+        for (int age = 1; age <= 5; age++) {
+            saveProfileWithAge("range-query-it-page-" + age, age);
+        }
+
+        refreshPersistence(Profile.class);
+
+        PartialList<Profile> firstPage = keepTrying(
+                "First range query page should be available",
+                () -> persistenceService.rangeQuery(AGE_PROPERTY, "1", "6", AGE_PROPERTY + ":asc", Profile.class, 0, 2),
+                r -> r != null && r.getList().size() == 2 && r.getTotalSize() == 5,
+                DEFAULT_TRYING_TIMEOUT, DEFAULT_TRYING_TRIES);
+
+        Assert.assertEquals(2, firstPage.getList().size());
+        Assert.assertEquals(5, firstPage.getTotalSize());
+        Assert.assertEquals(1, firstPage.getList().get(0).getProperty("age"));
+        Assert.assertEquals(2, firstPage.getList().get(1).getProperty("age"));
+
+        PartialList<Profile> lastPage = keepTrying(
+                "Last range query page should be available",
+                () -> persistenceService.rangeQuery(AGE_PROPERTY, "1", "6", AGE_PROPERTY + ":asc", Profile.class, 4, 2),
+                r -> r != null && r.getList().size() == 1 && r.getTotalSize() == 5,
+                DEFAULT_TRYING_TIMEOUT, DEFAULT_TRYING_TRIES);
+
+        Assert.assertEquals(1, lastPage.getList().size());
+        Assert.assertEquals(5, lastPage.getTotalSize());
+        Assert.assertEquals(5, lastPage.getList().get(0).getProperty("age"));
+    }
+
+    private void saveProfileWithAge(String idSuffix, int age) {
+        Profile profile = new Profile();
+        profile.setItemId(idSuffix + "-" + UUID.randomUUID());
+        profile.setProperty("age", age);
+        persistenceService.save(profile);
+        profileIds.add(profile.getItemId());
+    }
+}

persistence-elasticsearch/core/src/main/java/org/apache/unomi/persistence/elasticsearch/ElasticSearchPersistenceServiceImpl.java

@@ -2074,6 +2074,20 @@ private String getPropertyNameWithData(String name, String itemType) {
         return query(Query.of(q -> q.queryString(qs -> qs.query(fulltext))), sortBy, clazz, offset, size, null, null);
     }
 
+    @Override
+    public <T extends Item> PartialList<T> rangeQuery(String fieldName, String from, String to, String sortBy, Class<T> clazz, int offset, int size) {
+        return query(Query.of(q -> q.range(r -> r.untyped(v -> {
+            v.field(fieldName);
+            if (from != null) {
+                v.gte(JsonData.of(from));
+            }
+            if (to != null) {
+                v.lte(JsonData.of(to));
+            }
+            return v;
+        }))), sortBy, clazz, offset, size, null, null);
+    }
+
     @Override public long queryCount(Condition query, String itemType) {
         try {
             return conditionESQueryBuilderDispatcher.count(query);

persistence-opensearch/core/src/main/java/org/apache/unomi/persistence/opensearch/OpenSearchPersistenceServiceImpl.java

@@ -1948,6 +1948,20 @@ public <T extends Item> PartialList<T> queryFullText(String fulltext, String sor
         return query(Query.of(q->q.queryString(s->s.query(fulltext))), sortBy, clazz, offset, size, null, null);
     }
 
+    @Override
+    public <T extends Item> PartialList<T> rangeQuery(String fieldName, String from, String to, String sortBy, Class<T> clazz, int offset, int size) {
+        return query(Query.of(q -> q.range(r -> {
+            r.field(fieldName);
+            if (from != null) {
+                r.from(JsonData.of(from));
+            }
+            if (to != null) {
+                r.to(JsonData.of(to));
+            }
+            return r;
+        })), sortBy, clazz, offset, size, null, null);
+    }
+
     @Override
     public long queryCount(Condition query, String itemType) {
         try {

persistence-spi/src/main/java/org/apache/unomi/persistence/spi/PersistenceService.java

@@ -693,6 +693,27 @@ default <T extends Item> void refreshIndex(Class<T> clazz) {
      */
     <T extends Item> void purgeTimeBasedItems(int existsNumberOfDays, Class<T> clazz);
 
+    /**
+     * Retrieves all items of the specified Item subclass which specified ranged property is within the specified bounds, ordered according to the specified {@code sortBy} String
+     * and paged: only {@code size} of them are retrieved, starting with the {@code offset}-th one.
+     * <p>
+     * Both bounds are inclusive: items whose property value equals {@code from} or {@code to} are included in the results. Either bound may be {@code null} to leave that side
+     * of the range unbounded.
+     *
+     * @param <T>       the type of the Item subclass we want to retrieve
+     * @param fieldName the name of the range property we want items to retrieve to be included between the specified start and end points
+     * @param from      the beginning (inclusive) of the range we want to consider, or {@code null} for no lower bound
+     * @param to        the end (inclusive) of the range we want to consider, or {@code null} for no upper bound
+     * @param sortBy    an optional ({@code null} if no sorting is required) String of comma ({@code ,}) separated property names on which ordering should be performed, ordering
+     *                  elements according to the property order in the String, considering each in turn and moving on to the next one in case of equality of all preceding ones.
+     *                  Each property name is optionally followed by a column ({@code :}) and an order specifier: {@code asc} or {@code desc}.
+     * @param clazz     the {@link Item} subclass of the items we want to retrieve
+     * @param offset    zero or a positive integer specifying the position of the first item in the total ordered collection of matching items
+     * @param size      a positive integer specifying how many matching items should be retrieved or {@code -1} if all of them should be retrieved
+     * @return a {@link PartialList} of items matching the specified criteria
+     */
+    <T extends Item> PartialList<T> rangeQuery(String fieldName, String from, String to, String sortBy, Class<T> clazz, int offset, int size);
+
     /**
      * Retrieves the specified metrics for the specified field of items of the specified type as defined by the Item subclass public field {@code ITEM_TYPE} and matching the
      * specified {@link Condition}.

services/src/test/java/org/apache/unomi/services/impl/InMemoryPersistenceServiceImpl.java

@@ -968,9 +968,7 @@ public <T extends Item> PartialList<T> query(String fieldName, String fieldValue
         return createPartialList(items, offset, size);
     }
 
-    /**
-     * Test-harness helper for range queries. Not on PersistenceService on master yet.
-     */
+    @Override
     public <T extends Item> PartialList<T> rangeQuery(String fieldName, String from, String to, String sortBy, Class<T> clazz, int offset, int size) {
         List<T> items = filterItemsByClass(clazz);
         items = items.stream()

services/src/test/java/org/apache/unomi/services/impl/InMemoryPersistenceServiceImplTest.java

@@ -3140,6 +3140,7 @@ void shouldHandlePaginationInRangeQueries() {
 
             // then
             assertEquals(2, page1.getList().size());
+            assertEquals(5, page1.getTotalSize());
             assertEquals(1.0, page1.getList().get(0).getNumericValue());
             assertEquals(2.0, page1.getList().get(1).getNumericValue());
 
@@ -3152,6 +3153,7 @@ void shouldHandlePaginationInRangeQueries() {
 
             // then
             assertEquals(2, page2.getList().size());
+            assertEquals(5, page2.getTotalSize());
             assertEquals(3.0, page2.getList().get(0).getNumericValue());
             assertEquals(4.0, page2.getList().get(1).getNumericValue());
 
@@ -3164,6 +3166,7 @@ void shouldHandlePaginationInRangeQueries() {
 
             // then
             assertEquals(1, page3.getList().size());
+            assertEquals(5, page3.getTotalSize());
             assertEquals(5.0, page3.getList().get(0).getNumericValue());
         }
 

services/src/test/java/org/apache/unomi/services/impl/cluster/ClusterServiceImplTest.java

@@ -107,8 +107,12 @@ public void setUp() {
         // Set scheduler in cluster service - this would normally be done by OSGi but we need to do it manually in tests
         clusterService.setSchedulerService(schedulerService);
 
-        // Explicitly initialize scheduled tasks to handle the circular dependency properly
-        clusterService.initializeScheduledTasks();
+        // Note: scheduled tasks are intentionally NOT started here. clusterStaleNodesCleanup and
+        // clusterNodeStatisticsUpdate both have initialDelay=0, so starting them eagerly in every
+        // test's setUp() raced the test body on a real background thread pool and could delete
+        // freshly-saved fixtures before assertions ran (flaky only under CI-like scheduling/timing).
+        // Tests that actually exercise scheduled-task behavior call clusterService.init() themselves,
+        // which starts them at the right time.
     }
 
     @Test