Added the ability to batch load children objects and have their

resolution deferred until the batch completed. Also unit tests on the
same.

Author: tim-aero

README.md

@@ -1625,3 +1625,10 @@ When mapping a Java object to Aerospike the most common operations to do are to
 - Document creation of builder -- multiple configuration files are allowed, if the same class is declared in both the first one encountered wins. 
 - Document methods with 2 parameters for keys and setters, the second one either a Key or a Value
 - Document subclasses and the mapping to tables + references stored as lists
+- Batch load of child items on Maps and References. Ensure testing of non-parameterized classes too. Also of methods on Virtual LIsts
+- Test (and fix) batch loading of children on virtual lists.
+- Document batch loading
+- Ensure batchloading option exists in AerospikeReference Configuration
+- handle object graph circularities (A->B->C). Be careful of: A->B(Lazy), A->C->B: B should end up fully hydrated in both instances, not lazy in both instances
+- Consider the items on virtual list which return a list to be able to return a map as well (ELEMENT_LIST, ELEMENT_MAP) 
+- Test if map supports lazy loading of referenced objects.

src/main/java/com/aerospike/mapper/annotations/AerospikeReference.java

@@ -17,6 +17,12 @@
 	 * @return
 	 */
 	boolean lazy() default false;
+	/**
+	 * When a reference is to be loaded, it can either be loaded inline or it can be loaded via a batch load. The
+	 * batch load is typically significantly more efficient. Set this flag to <pre>false</pre> to prevent the batch load
+	 * @return
+	 */
+	boolean batchLoad() default true;
 
 	public static enum ReferenceType {
 		ID,

src/main/java/com/aerospike/mapper/tools/AeroMapper.java

@@ -26,6 +26,8 @@
 import com.aerospike.client.query.RecordSet;
 import com.aerospike.client.query.Statement;
 import com.aerospike.mapper.tools.ClassCache.PolicyType;
+import com.aerospike.mapper.tools.DeferredObjectLoader.DeferredObject;
+import com.aerospike.mapper.tools.DeferredObjectLoader.DeferredObjectSetter;
 import com.aerospike.mapper.tools.TypeUtils.AnnotatedType;
 import com.aerospike.mapper.tools.configuration.ClassConfig;
 import com.aerospike.mapper.tools.configuration.Configuration;
@@ -239,9 +241,11 @@ public Object translateToAerospike(Object obj) {
      * @return
      */
     @SuppressWarnings("unchecked")
-	public <T> T translateFromAerospike(Object obj, Class<T> expectedClazz) {
+	public <T> T translateFromAerospike(@NotNull Object obj, @NotNull Class<T> expectedClazz) {
     	TypeMapper thisMapper = TypeUtils.getMapper(expectedClazz, AnnotatedType.getDefaultAnnotateType(), this);
-    	return (T)(thisMapper == null ? obj : thisMapper.fromAerospikeFormat(obj));
+    	T result = (T)(thisMapper == null ? obj : thisMapper.fromAerospikeFormat(obj));
+		resolveDependencies(ClassCache.getInstance().loadClass(expectedClazz, this));
+		return result;
     }
 
 
@@ -276,34 +280,61 @@ public void update(@NotNull Object object, String ... binNames) throws Aerospike
         save(null, object, RecordExistsAction.UPDATE, binNames);
     }
 
-
     public <T> T readFromDigest(Policy readPolicy, @NotNull Class<T> clazz, @NotNull byte[] digest) throws AerospikeException {
+    	return this.readFromDigest(readPolicy, clazz, digest, true);
+    }
+    
+    /**
+     * This method should not be used except by mappers
+     */
+    public <T> T readFromDigest(Policy readPolicy, @NotNull Class<T> clazz, @NotNull byte[] digest, boolean resolveDependencies) throws AerospikeException {
         ClassCacheEntry<T> entry = getEntryAndValidateNamespace(clazz);
         Key key = new Key(entry.getNamespace(), digest, entry.getSetName(), null);
-        return this.read(readPolicy, clazz, key, entry);
+        return this.read(readPolicy, clazz, key, entry, resolveDependencies);
     }
 
     public <T> T readFromDigest(@NotNull Class<T> clazz, @NotNull byte[] digest) throws AerospikeException {
+    	return this.readFromDigest(clazz, digest, true);
+    }
+    
+    /**
+     * This method should not be used except by mappers
+     */
+    public <T> T readFromDigest(@NotNull Class<T> clazz, @NotNull byte[] digest, boolean resolveDependencies) throws AerospikeException {
         ClassCacheEntry<T> entry = getEntryAndValidateNamespace(clazz);
         Key key = new Key(entry.getNamespace(), digest, entry.getSetName(), null);
-        return this.read(null, clazz, key, entry);
+        return this.read(null, clazz, key, entry, resolveDependencies);
     }
 
     public <T> T read(Policy readPolicy, @NotNull Class<T> clazz, @NotNull Object userKey) throws AerospikeException {
+    	return this.read(readPolicy, clazz, userKey, true);
+    }
+    
+    /**
+     * This method should not be used except by mappers
+     */
+    public <T> T read(Policy readPolicy, @NotNull Class<T> clazz, @NotNull Object userKey, boolean resolveDependencies) throws AerospikeException {
         ClassCacheEntry<T> entry = getEntryAndValidateNamespace(clazz);
         String set = entry.getSetName();
         Key key = new Key(entry.getNamespace(), set, Value.get(entry.translateKeyToAerospikeKey(userKey)));
-        return read(readPolicy, clazz, key, entry);
+        return read(readPolicy, clazz, key, entry, resolveDependencies);
     }
 
     public <T> T read(@NotNull Class<T> clazz, @NotNull Object userKey) throws AerospikeException {
+    	return this.read(clazz, userKey, true);
+    }
+    
+    /**
+     * This method should not be used: It is used by mappers to correctly resolved dependencies. Use read(clazz, userkey) instead
+     */
+    public <T> T read(@NotNull Class<T> clazz, @NotNull Object userKey, boolean resolveDependencies) throws AerospikeException {
         ClassCacheEntry<T> entry = getEntryAndValidateNamespace(clazz);
         String set = entry.getSetName();
         Key key = new Key(entry.getNamespace(), set, Value.get(entry.translateKeyToAerospikeKey(userKey)));
-        return read(null, clazz, key, entry);
+        return read(null, clazz, key, entry, resolveDependencies);
     }
 
-    private <T> T read(Policy readPolicy, @NotNull Class<T> clazz, @NotNull Key key, @NotNull ClassCacheEntry<T> entry) {
+    private <T> T read(Policy readPolicy, @NotNull Class<T> clazz, @NotNull Key key, @NotNull ClassCacheEntry<T> entry, boolean resolveDepenencies) {
     	if (readPolicy == null) {
     		readPolicy = entry.getReadPolicy();
     	}
@@ -314,7 +345,7 @@ private <T> T read(Policy readPolicy, @NotNull Class<T> clazz, @NotNull Key key,
         } else {
             try {
             	ThreadLocalKeySaver.save(key);
-                T result = (T) convertToObject(clazz, record, entry);
+                T result = (T) convertToObject(clazz, record, entry, resolveDepenencies);
                 return result;
             } catch (ReflectiveOperationException e) {
                 throw new AerospikeException(e);
@@ -325,6 +356,7 @@ private <T> T read(Policy readPolicy, @NotNull Class<T> clazz, @NotNull Key key,
         }
     }
 
+    
     public <T> boolean delete(@NotNull Class<T> clazz, @NotNull Object userKey) throws AerospikeException {
     	return this.delete(null, clazz, userKey);
     }
@@ -461,18 +493,39 @@ public <T> T convertToObject(Class<T> clazz, Record record) {
     }
 
     public <T> T convertToObject(Class<T> clazz, Record record, ClassCacheEntry<T> entry) throws ReflectiveOperationException {
+    	return this.convertToObject(clazz, record, entry, true);
+    }
+    
+    /**
+     * This method should not be used, it is public only to allow mappers to see it.
+     */
+    public <T> T convertToObject(Class<T> clazz, Record record, ClassCacheEntry<T> entry, boolean resolveDependencies) throws ReflectiveOperationException {
         if (entry == null) {
             entry = ClassCache.getInstance().loadClass(clazz, this);
         }
-        return entry.constructAndHydrate(clazz, record);
+        T result = entry.constructAndHydrate(clazz, record);
+        if (resolveDependencies) {
+        	resolveDependencies(entry);
+        }
+		return result;
     }
 
     public <T> T convertToObject(Class<T> clazz, List<Object> record)  {
+    	return this.convertToObject(clazz, record, true);
+    }
+
+    /**
+     * This method should not be used, it is public only to allow mappers to see it.
+     */
+    public <T> T convertToObject(Class<T> clazz, List<Object> record, boolean resolveDependencies)  {
 		try {
 	        ClassCacheEntry<T> entry = ClassCache.getInstance().loadClass(clazz, this);
 	        T result;
 			result = clazz.getConstructor().newInstance();
 			entry.hydrateFromList(record, result);
+			if (resolveDependencies) {
+				resolveDependencies(entry);
+			}
 			return result;
 		} catch (ReflectiveOperationException e) {
 			throw new AerospikeException(e);
@@ -499,4 +552,74 @@ public <T> Map<String, Object> convertToMap(@NotNull T instance) {
     	ClassCacheEntry<T> entry = (ClassCacheEntry<T>) ClassCache.getInstance().loadClass(instance.getClass(), this);
     	return entry.getMap(instance, false);
     }
+    
+    /**
+     * If an object refers to other objects (eg A has a list of B via references), then reading the object will populate the
+     * ids. If configured to do so, these objects can be loaded via a batch load and populated back into the references which
+     * contain them. This method performs this batch load, translating the records to objects and mapping them back to the
+     * references.
+     * <p/>
+     * These loaded child objects can themselves have other references to other objects, so we iterate through this until
+     * the list of deferred objects is empty. The deferred objects are stored in a <pre>ThreadLocalData<pre> list, so are thread safe
+     * @param parentEntity - the ClassCacheEntry of the parent entity. This is used to get the batch policy to use.
+     */
+    private void resolveDependencies(ClassCacheEntry<?> parentEntity) {
+    	List<DeferredObjectSetter> deferredObjects = DeferredObjectLoader.getAndClear();
+    	
+    	if (deferredObjects.size() == 0) {
+    		return;
+    	}
+    	
+    	BatchPolicy batchPolicy = parentEntity == null ? mClient.getBatchPolicyDefault() : parentEntity.getBatchPolicy();
+    	BatchPolicy batchPolicyClone = new BatchPolicy(batchPolicy);
+    	
+    	while (deferredObjects != null && !deferredObjects.isEmpty()) {
+    		int size = deferredObjects.size();
+    		
+    		ClassCacheEntry<?>[] classCaches = new ClassCacheEntry<?>[size];
+    		Key[] keys = new Key[size];
+    		
+    		for (int i = 0; i < size; i++) {
+    			DeferredObjectSetter thisObjectSetter = deferredObjects.get(i);
+    			DeferredObject deferredObject = thisObjectSetter.getObject();
+    			Class<?> clazz = deferredObject.getType();
+    			ClassCacheEntry<?> entry = (ClassCacheEntry<?>) ClassCache.getInstance().loadClass(clazz, this);
+    			classCaches[i] = entry; 
+    			
+    			if (deferredObject.isDigest()) {
+    				keys[i] = new Key(entry.getNamespace(), (byte[])deferredObject.getKey(), entry.getSetName(), null);
+    			}
+    			else {
+    				keys[i] = new Key(entry.getNamespace(), entry.getSetName(), Value.get(entry.translateKeyToAerospikeKey(deferredObject.getKey())));
+    			}
+    		}
+    		
+    		// Load the data
+    		if (keys.length <= 2) {
+    			// Just single-thread these keys for speed
+    			batchPolicyClone.maxConcurrentThreads = 1;
+    		}
+    		else {
+    			batchPolicyClone.maxConcurrentThreads = batchPolicy.maxConcurrentThreads;
+    		}
+    		Record[] records = this.mClient.get(batchPolicyClone, keys);
+    		
+    		for (int i = 0; i < size; i++) {
+    			DeferredObjectSetter thisObjectSetter = deferredObjects.get(i);
+    			try {
+                	ThreadLocalKeySaver.save(keys[i]);
+                	Object result = this.convertToObject((Class)thisObjectSetter.getObject().getType(), records[i], classCaches[i], false);
+                	thisObjectSetter.getSetter().setValue(result);
+	            } catch (ReflectiveOperationException e) {
+	                throw new AerospikeException(e);
+	            }
+	            finally {
+	            	ThreadLocalKeySaver.clear();
+	            }
+    		}
+        	deferredObjects = DeferredObjectLoader.getAndClear();
+    	}
+    }
+    
+
 }

src/main/java/com/aerospike/mapper/tools/ClassCache.java

@@ -45,6 +45,9 @@ private ClassCache() {
 	}
 
 	public <T> ClassCacheEntry<T> loadClass(@NotNull Class<T> clazz, AeroMapper mapper) {
+		if (clazz.isPrimitive() || clazz.equals(Object.class) || clazz.equals(String.class) || Number.class.isAssignableFrom(clazz)) {
+			return null;
+		}
 		ClassCacheEntry<T> entry = cacheMap.get(clazz);
 		if (entry == null) {
 			try {

src/main/java/com/aerospike/mapper/tools/ClassCacheEntry.java

@@ -20,7 +20,9 @@
 
 import com.aerospike.client.AerospikeException;
 import com.aerospike.client.Bin;
+import com.aerospike.client.Key;
 import com.aerospike.client.Record;
+import com.aerospike.client.Value;
 import com.aerospike.client.cdt.MapOrder;
 import com.aerospike.client.policy.BatchPolicy;
 import com.aerospike.client.policy.Policy;
@@ -36,6 +38,8 @@
 import com.aerospike.mapper.annotations.AerospikeRecord;
 import com.aerospike.mapper.annotations.AerospikeSetter;
 import com.aerospike.mapper.annotations.ParamFrom;
+import com.aerospike.mapper.tools.DeferredObjectLoader.DeferredObject;
+import com.aerospike.mapper.tools.DeferredObjectLoader.DeferredObjectSetter;
 import com.aerospike.mapper.tools.TypeUtils.AnnotatedType;
 import com.aerospike.mapper.tools.configuration.BinConfig;
 import com.aerospike.mapper.tools.configuration.ClassConfig;
@@ -137,6 +141,15 @@ public Policy getReadPolicy() {
 	public WritePolicy getWritePolicy() {
 		return writePolicy;
 	}
+	public BatchPolicy getBatchPolicy() {
+		return batchPolicy;
+	}
+	public QueryPolicy getQueryPolicy() {
+		return queryPolicy;
+	}
+	public ScanPolicy getScanPolicy() {
+		return scanPolicy;
+	}
 
 	public Class<?> getUnderlyingClass() {
 		return this.clazz;
@@ -752,6 +765,7 @@ private T constructAndHydrate(Class<T> clazz, Record record, Map<String, Object>
 				valueMap.clear();
 				thisClass = thisClass.superClazz;
 			}
+			
 			return result;
 		}
 		catch (ReflectiveOperationException ref) {
@@ -909,7 +923,7 @@ public void hydrateFromList(List<Object> list, Object instance, boolean skipKey)
 		try {
 			int index = 0;
 			int endIndex = list.size();
-			ClassCacheEntry thisClass = this;
+			ClassCacheEntry<?> thisClass = this;
 			while (thisClass != null) {
 				if (index < endIndex) {
 					Object lastValue = list.get(endIndex-1);

src/main/java/com/aerospike/mapper/tools/DeferredObjectLoader.java

@@ -0,0 +1,79 @@
+package com.aerospike.mapper.tools;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class DeferredObjectLoader {
+	public static interface DeferredSetter {
+		public void setValue(Object object);
+	}
+	
+	public static class DeferredObject {
+		private final Object key;
+		private final Class<?> type;
+		private final boolean isDigest;
+		public DeferredObject(Object key, Class<?> type, boolean isDigest) {
+			super();
+			this.key = key;
+			this.type = type;
+			this.isDigest = isDigest;
+		}
+		public Object getKey() {
+			return key;
+		}
+		public Class<?> getType() {
+			return type;
+		}
+		public boolean isDigest() {
+			return isDigest;
+		}
+	}
+	
+	public static class DeferredObjectSetter {
+		private final DeferredSetter setter;
+		private final DeferredObject object;
+		
+		public DeferredObjectSetter(DeferredSetter setter, DeferredObject object) {
+			super();
+			this.setter = setter;
+			this.object = object;
+		}
+		public DeferredSetter getSetter() {
+			return setter;
+		}
+		public DeferredObject getObject() {
+			return object;
+		}
+	}
+	
+	
+	private static ThreadLocal<List<DeferredObjectSetter>> threadLocalLoader = new ThreadLocal<List<DeferredObjectSetter>>() {
+		@Override
+		public List<DeferredObjectSetter> initialValue() {
+			return new ArrayList<DeferredObjectSetter>();
+		}
+	};
+	
+	public static void save(DeferredObjectSetter object) {
+		threadLocalLoader.get().add(object);
+	}
+	
+	public static void clear() {
+		threadLocalLoader.get().clear();
+	}
+	
+	public static List<DeferredObjectSetter> get() {
+		return threadLocalLoader.get();
+	}
+	
+	public static void add(DeferredObjectSetter deferredSetter) {
+		threadLocalLoader.get().add(deferredSetter);
+	}
+
+	public static List<DeferredObjectSetter> getAndClear() {
+		List<DeferredObjectSetter> localArray = threadLocalLoader.get();
+		List<DeferredObjectSetter> setters = new ArrayList<DeferredObjectSetter>(localArray);
+		localArray.clear();
+		return setters;
+	}
+}