Added Scan and Query functionality (#68)

tfaulkes · web-flow · commit 0dc4ba3acb93 · 2021-08-23T18:33:32.000+03:00
* Added Scan and Query functionality
diff --git a/README.md b/README.md
@@ -38,6 +38,8 @@ The documentation for this project can be found on [javadoc.io](https://www.java
         + 10.1.3. [Embed Structure](#Embed-Structure)
         + 10.1.4. [Reference Structure](#Reference-Structure)
 11. [Virtual Lists](#Virtual-Lists)
+12. [Scans](#Scans)
+13. [Queries](#Queries)
 
 # Compatibility with Aerospike Clients
 
@@ -2168,3 +2170,98 @@ public <T> T convertToObject(Class<T> clazz, Record record);
 
 Note: At the moment not all CDT operations are supported, and if the underlying CDTs are of the wrong type, a different API call may be used. For example, if you invoke `getByKeyRange` on items represented in the database as a list, `getByValueRange` is invoked instead as a list has no key.
 
+## Scans
+Scans can be used to process every record in a set. The scan iterates through every item in the set and invokes a callback for every item in the set. For example:
+
+```java
+mapper.scan(Pseron.class, (person) -> {
+	// ... process person
+	return true;
+});
+```
+
+If the processing method returns true, the scan continues. However, if the processing method returns false the scan will abort. Note that if the scan policy calls for multi-threading of the scans, the callback method may be invoked by multiple threads at once and hence must be thread safe. If one thread aborts the scan, other threads already in the processing method will finish processing their records.
+
+Note that if you want to process only some records in a set you can attach an Expression on the optional policy passed to the scan. For example, if there is a `Person` class:
+
+```java
+@AerospikeRecord(namespace = "test", set = "testScan")
+public class Person {
+	@AerospikeKey
+	private int id;
+	private String name;
+	private int age;
+	
+	public Person(@ParamFrom("id") int id, @ParamFrom("name") String name, @ParamFrom("age") int age) {
+		super();
+		this.id = id;
+		this.name = name;
+		this.age = age;
+	}
+
+	public int getId() {
+		return id;
+	}
+
+	public String getName() {
+		return name;
+	}
+
+	public int getAge() {
+		return age;
+	}
+}
+```
+ 
+and then several people are inserted:
+ 
+ ```java
+mapper.save(new Person(1, "Tim", 312),
+			new Person(2, "Bob", 44),
+			new Person(3, "Sue", 56),
+			new Person(4, "Rob", 23),
+			new Person(5, "Jim", 32),
+			new Person(6, "Bob", 78));
+ ```
+
+As a contrived example, let's say we want to count the number of people in the set called "Bob". We can simply do:
+
+```java
+AtomicInteger counter = new AtomicInteger(0);
+ScanPolicy scanPolicy = new ScanPolicy(mapper.getScanPolicy(Person.class));
+scanPolicy.filterExp = Exp.build(Exp.eq(Exp.stringBin("name"), Exp.val("Bob")));
+mapper.scan(scanPolicy, Person.class, (person) -> {
+	counter.incrementAndGet();
+	return true;
+});
+```
+
+Note that when we altered the ScanPolicy, we had to make a copy of it first. If we fail to do this, the ScanPolicy will be altered for all subsequent calls. To clarify, the **wrong** way to set the scan policy is
+
+```java
+ScanPolicy scanPolicy = mapper.getScanPolicy(Person.class);
+scanPolicy.filterExp = Exp.build(Exp.eq(Exp.stringBin("name"), Exp.val("Bob")));
+```
+
+and the **right** way to set an expression is
+
+```java
+ScanPolicy scanPolicy = new ScanPolicy(mapper.getScanPolicy(Person.class));
+scanPolicy.filterExp = Exp.build(Exp.eq(Exp.stringBin("name"), Exp.val("Bob")));
+```
+
+## Queries
+
+Similar to Scans, Queries can processed using the AeroMapper. Syntactically, the only difference between a query and a scan is the addition of a `Filter` on the Query which dictates the criteria of the query. A secondary index must be defined on the Bin referenced in the Filter or an error will be thrown. If no filter is passed, the query will be turned into a scan.
+
+Similar to Scans, returning `false` on the processing method will abort the Query and process no further records, and additional filter criteria can be added using Expressions on the QueryPolicy.
+
+```java
+mapper.query(A.class, (a) -> {
+	System.out.println(a);
+	counter.incrementAndGet();
+	return true;
+}, Filter.range("age", 30, 54));
+
+```
+
diff --git a/src/main/java/com/aerospike/mapper/tools/AeroMapper.java b/src/main/java/com/aerospike/mapper/tools/AeroMapper.java
@@ -6,13 +6,15 @@
 import java.lang.reflect.Array;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 
 import javax.validation.constraints.NotNull;
 
 import org.apache.commons.lang3.StringUtils;
 
 import com.aerospike.client.AerospikeException;
+import com.aerospike.client.AerospikeException.ScanTerminated;
 import com.aerospike.client.Bin;
 import com.aerospike.client.IAerospikeClient;
 import com.aerospike.client.Key;
@@ -24,6 +26,7 @@
 import com.aerospike.client.policy.RecordExistsAction;
 import com.aerospike.client.policy.ScanPolicy;
 import com.aerospike.client.policy.WritePolicy;
+import com.aerospike.client.query.Filter;
 import com.aerospike.client.query.RecordSet;
 import com.aerospike.client.query.Statement;
 import com.aerospike.mapper.tools.ClassCache.PolicyType;
@@ -613,6 +616,153 @@ public <T> void find(@NotNull Class<T> clazz, Function<T, Boolean> function) thr
         }
     }
 
+    /**
+     * Scan every record in the set associated with the passed class. Each record will be converted to the appropriate class then passed to the
+     * processor. If the processor returns true, more records will be processed and if the processor returns false, the scan is aborted.
+     * <p/>
+     * Depending on the ScanPolicy set up for this class, it is possible for the processor to be called by multiple different
+     * threads concurrently, so the processor should be thread-safe
+     *
+     * @param clazz     - the class used to determine which set to scan and to convert the returned records to.
+     * @param processor - the Processor used to process each record
+     */
+    @Override
+    public <T> void scan(@NotNull Class<T> clazz, @NotNull Processor<T> processor) {
+        scan(null, clazz, processor);
+    }
+
+    /**
+     * Scan every record in the set associated with the passed class. Each record will be converted to the appropriate class then passed to the
+     * processor. If the processor returns true, more records will be processed and if the processor returns false, the scan is aborted.
+     * <p/>
+     * Depending on the policy passed or set as the ScanPolicy for this class, it is possible for the processor to be called by multiple different
+     * threads concurrently, so the processor should be thread-safe. Note that as a consequence of this, if the processor returns false to abort the
+     * scan there is a chance that records are being concurrently processed in other threads and this processing will not be interrupted.
+     * <p/>
+     *
+     * @param policy    - the scan policy to use. If this is null, the default scan policy of the passed class will be used.
+     * @param clazz     - the class used to determine which set to scan and to convert the returned records to.
+     * @param processor - the Processor used to process each record
+     */
+    @Override
+    public <T> void scan(ScanPolicy policy, @NotNull Class<T> clazz, @NotNull Processor<T> processor) {
+        scan(policy, clazz, processor, -1);
+    }
+
+    /**
+     * Scan every record in the set associated with the passed class, limiting the throughput to the specified recordsPerSecond. Each record will be converted
+     * to the appropriate class then passed to the
+     * processor. If the processor returns true, more records will be processed and if the processor returns false, the scan is aborted.
+     * <p/>
+     * Depending on the ScanPolicy set up for this class, it is possible for the processor to be called by multiple different
+     * threads concurrently, so the processor should be thread-safe
+     *
+     * @param clazz            - the class used to determine which set to scan and to convert the returned records to.
+     * @param processor        - the Processor used to process each record
+     * @param recordsPerSecond - the maximum number of records to be processed every second.
+     */
+    @Override
+    public <T> void scan(@NotNull Class<T> clazz, @NotNull Processor<T> processor, int recordsPerSecond) {
+        scan(null, clazz, processor, recordsPerSecond);
+    }
+
+    /**
+     * Scan every record in the set associated with the passed class. Each record will be converted to the appropriate class then passed to the
+     * processor. If the processor returns true, more records will be processed and if the processor returns false, the scan is aborted.
+     * <p/>
+     * Depending on the policy passed or set as the ScanPolicy for this class, it is possible for the processor to be called by multiple different
+     * threads concurrently, so the processor should be thread-safe. Note that as a consequence of this, if the processor returns false to abort the
+     * scan there is a chance that records are being concurrently processed in other threads and this processing will not be interrupted.
+     * <p/>
+     *
+     * @param policy           - the scan policy to use. If this is null, the default scan policy of the passed class will be used.
+     * @param clazz            - the class used to determine which set to scan and to convert the returned records to.
+     * @param processor        - the Processor used to process each record
+     * @param recordsPerSecond - the number of records to process per second. Set to 0 for unlimited, &gt; 0 for a finite rate, &lt; 0 for no change
+     *                         (use the value from the passed policy)
+     */
+    @Override
+    public <T> void scan(ScanPolicy policy, @NotNull Class<T> clazz, @NotNull Processor<T> processor, int recordsPerSecond) {
+        ClassCacheEntry<T> entry = MapperUtils.getEntryAndValidateNamespace(clazz, this);
+        if (policy == null) {
+            policy = entry.getScanPolicy();
+        }
+        if (recordsPerSecond >= 0) {
+            // Ensure the underlying rate on the policy does not change
+            policy = new ScanPolicy(policy);
+            policy.recordsPerSecond = recordsPerSecond;
+        }
+        String namespace = entry.getNamespace();
+        String setName = entry.getSetName();
+
+        AtomicBoolean userTerminated = new AtomicBoolean(false);
+        try {
+            mClient.scanAll(policy, namespace, setName, (key, record) -> {
+                T object = this.getMappingConverter().convertToObject(clazz, record);
+                if (!processor.process(object)) {
+                    userTerminated.set(true);
+                    throw new AerospikeException.ScanTerminated();
+                }
+            });
+        } catch (ScanTerminated st) {
+            if (!userTerminated.get()) {
+                throw st;
+            }
+        }
+    }
+
+    /**
+     * Perform a secondary index query with the specified query policy. Each record will be converted
+     * to the appropriate class then passed to the processor. If the processor returns false the query is aborted
+     * whereas if the processor returns true subsequent records (if any) are processed.
+     * <p/>
+     * The query policy used will be the one associated with the passed classtype.
+     *
+     * @param clazz     - the class used to determine which set to scan and to convert the returned records to.
+     * @param processor - the Processor used to process each record
+     * @param filter    - the filter used to determine which secondary index to use. If this filter is null, every record in the set
+     *                  associated with the passed classtype will be scanned, effectively turning the query into a scan
+     */
+    @Override
+    public <T> void query(@NotNull Class<T> clazz, @NotNull Processor<T> processor, Filter filter) {
+        query(null, clazz, processor, filter);
+    }
+
+    /**
+     * Perform a secondary index query with the specified query policy. Each record will be converted
+     * to the appropriate class then passed to the processor. If the processor returns false the query is aborted
+     * whereas if the processor returns true subsequent records (if any) are processed.
+     *
+     * @param policy    - The query policy to use. If this parameter is not passed, the query policy associated with the passed classtype will be used
+     * @param clazz     - the class used to determine which set to scan and to convert the returned records to.
+     * @param processor - the Processor used to process each record
+     * @param filter    - the filter used to determine which secondary index to use. If this filter is null, every record in the set
+     *                  associated with the passed classtype will be scanned, effectively turning the query into a scan
+     */
+    @Override
+    public <T> void query(QueryPolicy policy, @NotNull Class<T> clazz, @NotNull Processor<T> processor, Filter filter) {
+        ClassCacheEntry<T> entry = MapperUtils.getEntryAndValidateNamespace(clazz, this);
+        if (policy == null) {
+            policy = entry.getQueryPolicy();
+        }
+        Statement statement = new Statement();
+        statement.setFilter(filter);
+        statement.setNamespace(entry.getNamespace());
+        statement.setSetName(entry.getSetName());
+
+        RecordSet recordSet = mClient.query(policy, statement);
+        try {
+            while (recordSet.next()) {
+                T object = this.getMappingConverter().convertToObject(clazz, recordSet.getRecord());
+                if (!processor.process(object)) {
+                    break;
+                }
+            }
+        } finally {
+            recordSet.close();
+        }
+    }
+    
     @Override
     public IAerospikeClient getClient() {
         return this.mClient;
diff --git a/src/main/java/com/aerospike/mapper/tools/IAeroMapper.java b/src/main/java/com/aerospike/mapper/tools/IAeroMapper.java
@@ -1,14 +1,18 @@
 package com.aerospike.mapper.tools;
 
+import java.util.function.Function;
+
+import javax.validation.constraints.NotNull;
+
 import com.aerospike.client.IAerospikeClient;
 import com.aerospike.client.policy.BatchPolicy;
 import com.aerospike.client.policy.Policy;
+import com.aerospike.client.policy.QueryPolicy;
+import com.aerospike.client.policy.ScanPolicy;
 import com.aerospike.client.policy.WritePolicy;
+import com.aerospike.client.query.Filter;
 import com.aerospike.mapper.tools.virtuallist.VirtualList;
 
-import javax.validation.constraints.NotNull;
-import java.util.function.Function;
-
 public interface IAeroMapper extends IBaseAeroMapper {
 
     void save(@NotNull Object... objects);
@@ -54,4 +58,16 @@ public interface IAeroMapper extends IBaseAeroMapper {
     <T> void find(@NotNull Class<T> clazz, Function<T, Boolean> function);
 
     IAerospikeClient getClient();
+
+    <T> void query(@NotNull Class<T> clazz, @NotNull Processor<T> processor, Filter filter);
+
+    <T> void query(QueryPolicy policy, @NotNull Class<T> clazz, @NotNull Processor<T> processor, Filter filter);
+
+    <T> void scan(@NotNull Class<T> clazz, @NotNull Processor<T> processor);
+
+    <T> void scan(ScanPolicy policy, @NotNull Class<T> clazz, @NotNull Processor<T> processor);
+
+    <T> void scan(@NotNull Class<T> clazz, @NotNull Processor<T> processor, int recordsPerSecond);
+
+    <T> void scan(ScanPolicy policy, @NotNull Class<T> clazz, @NotNull Processor<T> processor, int recordsPerSecond);
 }
diff --git a/src/main/java/com/aerospike/mapper/tools/Processor.java b/src/main/java/com/aerospike/mapper/tools/Processor.java
@@ -0,0 +1,11 @@
+package com.aerospike.mapper.tools;
+
+public interface Processor<T> {
+    /**
+     * Process the given record.
+     *
+     * @param data - the record to be processed
+     * @return true if further records should be processed, false if the processing loop should abort.
+     */
+    boolean process(T data);
+}
diff --git a/src/test/java/com/aerospike/mapper/QueryTest.java b/src/test/java/com/aerospike/mapper/QueryTest.java
diff --git a/src/test/java/com/aerospike/mapper/ScanTest.java b/src/test/java/com/aerospike/mapper/ScanTest.java
