FMWK-301 Support beforeLastUpdate parameter in deleteAll

src/main/java/org/springframework/data/aerospike/core/AerospikeTemplate.java

@@ -51,8 +51,10 @@
 import org.springframework.data.util.StreamUtils;
 import org.springframework.util.Assert;
 
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Calendar;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
@@ -538,15 +540,42 @@ private void deleteGroupedEntitiesByGroupedKeys(GroupedKeys groupedKeys) {
     @Override
     public <T> void deleteAll(Class<T> entityClass) {
         Assert.notNull(entityClass, "Class must not be null!");
-        deleteAll(getSetName(entityClass));
+        deleteAll(entityClass, (Long) null);
+    }
+
+    @Override
+    public <T> void deleteAll(Class<T> entityClass, Long beforeLastUpdateMillis) {
+        Assert.notNull(entityClass, "Class must not be null!");
+        deleteAll(getSetName(entityClass), beforeLastUpdateMillis);
+    }
+
+    @Override
+    public <T> void deleteAll(Class<T> entityClass, Instant beforeLastUpdate) {
+        Assert.notNull(entityClass, "Class must not be null!");
+        deleteAll(getSetName(entityClass), beforeLastUpdate);
     }
 
     @Override
     public void deleteAll(String setName) {
         Assert.notNull(setName, "Set name must not be null!");
+        deleteAll(setName, (Long) null);
+    }
+
+    @Override
+    public void deleteAll(String setName, Long beforeLastUpdateMillis) {
+        Assert.notNull(setName, "Set name must not be null!");
+        Instant beforeLastUpdateInstant = convertToInstant(beforeLastUpdateMillis);
+
+        deleteAll(setName, beforeLastUpdateInstant);
+    }
+
+    @Override
+    public void deleteAll(String setName, Instant beforeLastUpdate) {
+        Assert.notNull(setName, "Set name must not be null!");
+        Calendar beforeLastUpdateCalendar = convertToCalendar(beforeLastUpdate);
 
         try {
-            client.truncate(null, getNamespace(), setName, null);
+            client.truncate(null, getNamespace(), setName, beforeLastUpdateCalendar);
         } catch (AerospikeException e) {
             throw translateError(e);
         }

src/main/java/org/springframework/data/aerospike/core/BaseAerospikeTemplate.java

@@ -54,8 +54,11 @@
 import org.springframework.data.mapping.model.ConvertingPropertyAccessor;
 import org.springframework.util.Assert;
 
+import java.time.Instant;
+import java.util.Calendar;
 import java.util.Collection;
 import java.util.Comparator;
+import java.util.Date;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -391,6 +394,26 @@ private <S> S convertIfNecessary(Object source, Class<S> type) {
             : converter.getConversionService().convert(source, type);
     }
 
+    protected Instant convertToInstant(Long millis) {
+        if (millis == null) return null;
+
+        if (millis >= Instant.now().toEpochMilli())
+            throw new IllegalArgumentException("Last update time (%d) must be less than the current time"
+                .formatted(millis));
+        return Instant.ofEpochMilli(millis);
+    }
+
+    protected Calendar convertToCalendar(Instant instant) {
+        if (instant == null) return null;
+
+        Calendar calendar = Calendar.getInstance();
+        if (instant.toEpochMilli() > calendar.getTimeInMillis())
+            throw new IllegalArgumentException("Last update time (%d) must be less than the current time"
+                .formatted(instant.toEpochMilli()));
+        calendar.setTime(Date.from(instant));
+        return calendar;
+    }
+
     protected Operation[] getPutAndGetHeaderOperations(AerospikeWriteData data, boolean firstlyDeleteBins) {
         Bin[] bins = data.getBinsAsArray();
 

src/main/java/org/springframework/data/aerospike/core/ReactiveAerospikeOperations.java

@@ -34,6 +34,7 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.time.Instant;
 import java.util.Collection;
 import java.util.Map;
 import java.util.function.Supplier;
@@ -507,6 +508,26 @@ public interface ReactiveAerospikeOperations {
      */
     <T> Mono<Void> deleteAll(Class<T> entityClass);
 
+    /**
+     * Reactively truncate/delete all records in the set determined by the given entity class.
+     *
+     * @param entityClass            The class to extract set name from. Must not be {@literal null}.
+     * @param beforeLastUpdateMillis Delete records before the specified time (must be earlier than the current time at
+     *                               millisecond resolution). In UTC milliseconds from the epoch.
+     * @throws DataAccessException If operation failed (see {@link DefaultAerospikeExceptionTranslator} for details).
+     */
+    <T> Mono<Void> deleteAll(Class<T> entityClass, Long beforeLastUpdateMillis);
+
+    /**
+     * Reactively truncate/delete all records in the set determined by the given entity class.
+     *
+     * @param entityClass      The class to extract set name from. Must not be {@literal null}.
+     * @param beforeLastUpdate Delete records before the specified time (must be earlier than the current time at
+     *                         millisecond resolution).
+     * @throws DataAccessException If operation failed (see {@link DefaultAerospikeExceptionTranslator} for details).
+     */
+    <T> Mono<Void> deleteAll(Class<T> entityClass, Instant beforeLastUpdate);
+
     /**
      * Reactively truncate/delete all the documents in the given set.
      *
@@ -515,6 +536,26 @@ public interface ReactiveAerospikeOperations {
      */
     Mono<Void> deleteAll(String setName);
 
+    /**
+     * Reactively truncate/delete all documents in the given set.
+     *
+     * @param setName                Set name to truncate/delete all records in.
+     * @param beforeLastUpdateMillis Delete records before the specified time (must be earlier than the current time at
+     *                               millisecond resolution). In UTC milliseconds from the epoch.
+     * @throws DataAccessException If operation failed (see {@link DefaultAerospikeExceptionTranslator} for details).
+     */
+    Mono<Void> deleteAll(String setName, Long beforeLastUpdateMillis);
+
+    /**
+     * Reactively truncate/delete all documents in the given set.
+     *
+     * @param setName          Set name to truncate/delete all records in.
+     * @param beforeLastUpdate Delete records before the specified time (must be earlier than the current time at
+     *                         millisecond resolution).
+     * @throws DataAccessException If operation failed (see {@link DefaultAerospikeExceptionTranslator} for details).
+     */
+    Mono<Void> deleteAll(String setName, Instant beforeLastUpdate);
+
     /**
      * Find an existing record matching the document's class and id, add map values to the corresponding bins of the
      * record and return the modified record mapped to the document's class.

src/main/java/org/springframework/data/aerospike/core/ReactiveAerospikeTemplate.java

@@ -49,8 +49,10 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Calendar;
 import java.util.Collection;
 import java.util.Comparator;
 import java.util.List;
@@ -548,16 +550,46 @@ private Mono<Void> deleteEntitiesByGroupedKeys(GroupedKeys groupedKeys) {
     public <T> Mono<Void> deleteAll(Class<T> entityClass) {
         Assert.notNull(entityClass, "Class must not be null!");
 
-        return deleteAll(getSetName(entityClass));
+        return deleteAll(getSetName(entityClass), (Long) null);
+    }
+
+    @Override
+    public <T> Mono<Void> deleteAll(Class<T> entityClass, Long beforeLastUpdateMillis) {
+        Assert.notNull(entityClass, "Class must not be null!");
+
+        return deleteAll(getSetName(entityClass), beforeLastUpdateMillis);
+    }
+
+    @Override
+    public <T> Mono<Void> deleteAll(Class<T> entityClass, Instant beforeLastUpdate) {
+        Assert.notNull(entityClass, "Class must not be null!");
+
+        return deleteAll(getSetName(entityClass), beforeLastUpdate);
     }
 
     @Override
     public Mono<Void> deleteAll(String setName) {
         Assert.notNull(setName, "Set name must not be null!");
 
+        return deleteAll(setName, (Long) null);
+    }
+
+    @Override
+    public Mono<Void> deleteAll(String setName, Long beforeLastUpdateMillis) {
+        Assert.notNull(setName, "Set name must not be null!");
+        Instant beforeLastUpdateInstant = convertToInstant(beforeLastUpdateMillis);
+
+        return deleteAll(setName, beforeLastUpdateInstant);
+    }
+
+    @Override
+    public Mono<Void> deleteAll(String setName, Instant beforeLastUpdate) {
+        Assert.notNull(setName, "Set name must not be null!");
+        Calendar beforeLastUpdateCalendar = convertToCalendar(beforeLastUpdate);
+
         try {
             return Mono.fromRunnable(
-                () -> reactorClient.getAerospikeClient().truncate(null, namespace, setName, null));
+                () -> reactorClient.getAerospikeClient().truncate(null, namespace, setName, beforeLastUpdateCalendar));
         } catch (AerospikeException e) {
             throw translateError(e);
         }

src/test/java/org/springframework/data/aerospike/BaseBlockingIntegrationTests.java

@@ -6,12 +6,18 @@
 import org.springframework.data.aerospike.config.BlockingTestConfig;
 import org.springframework.data.aerospike.config.CommonTestConfig;
 import org.springframework.data.aerospike.core.AerospikeTemplate;
+import org.springframework.data.aerospike.query.FilterOperation;
+import org.springframework.data.aerospike.query.Qualifier;
 import org.springframework.data.aerospike.query.QueryEngine;
 import org.springframework.data.aerospike.query.cache.IndexRefresher;
 import org.springframework.data.aerospike.query.cache.IndexesCache;
+import org.springframework.data.aerospike.repository.query.Query;
 import org.springframework.data.aerospike.server.version.ServerVersionSupport;
 
 import java.util.Collection;
+import java.util.List;
+
+import static org.springframework.data.aerospike.repository.query.CriteriaDefinition.AerospikeMetadata.LAST_UPDATE_TIME;
 
 @SpringBootTest(
     classes = {BlockingTestConfig.class, CommonTestConfig.class},
@@ -43,4 +49,14 @@ protected <T> void deleteOneByOne(Collection<T> collection) {
     protected <T> void deleteOneByOne(Collection<T> collection, String setName) {
         collection.forEach(item -> template.delete(item, setName));
     }
+
+    protected <T> List<T> runLastUpdateTimeQuery(long lastUpdateTimeMillis, FilterOperation operation,
+                                                 Class<T> entityClass) {
+        Qualifier lastUpdateTimeLtMillis = Qualifier.metadataBuilder()
+            .setMetadataField(LAST_UPDATE_TIME)
+            .setFilterOperation(operation)
+            .setValue1AsObj(lastUpdateTimeMillis * 1000000)
+            .build();
+        return template.find(new Query(lastUpdateTimeLtMillis), entityClass).toList();
+    }
 }

src/test/java/org/springframework/data/aerospike/BaseReactiveIntegrationTests.java

@@ -6,11 +6,17 @@
 import org.springframework.data.aerospike.config.CommonTestConfig;
 import org.springframework.data.aerospike.config.ReactiveTestConfig;
 import org.springframework.data.aerospike.core.ReactiveAerospikeTemplate;
+import org.springframework.data.aerospike.query.FilterOperation;
+import org.springframework.data.aerospike.query.Qualifier;
 import org.springframework.data.aerospike.query.cache.ReactorIndexRefresher;
+import org.springframework.data.aerospike.repository.query.Query;
 import org.springframework.data.aerospike.server.version.ServerVersionSupport;
 import reactor.core.publisher.Flux;
 
 import java.io.Serializable;
+import java.util.List;
+
+import static org.springframework.data.aerospike.repository.query.CriteriaDefinition.AerospikeMetadata.LAST_UPDATE_TIME;
 
 @SpringBootTest(
     classes = {ReactiveTestConfig.class, CommonTestConfig.class},
@@ -46,4 +52,14 @@ protected <T> void deleteAll(Iterable<T> iterable) {
     protected <T> void deleteAll(Iterable<T> iterable, String setName) {
         Flux.fromIterable(iterable).flatMap(item -> reactiveTemplate.delete(item, setName)).blockLast();
     }
+
+    protected <T> List<T> runLastUpdateTimeQuery(long lastUpdateTimeMillis, FilterOperation operation,
+                                                 Class<T> entityClass) {
+        Qualifier lastUpdateTimeLtMillis = Qualifier.metadataBuilder()
+            .setMetadataField(LAST_UPDATE_TIME)
+            .setFilterOperation(operation)
+            .setValue1AsObj(lastUpdateTimeMillis * 1000000)
+            .build();
+        return reactiveTemplate.find(new Query(lastUpdateTimeLtMillis), entityClass).collectList().block();
+    }
 }