FMWK-468 Allow send key classes to use PK instead of Bin (#160)

* Allowed the use of the Aerospike Key instead of re-writing a key field.

Added a new annotation field "storeInPkOnly" to use the PK field as
storing the key, rather than explicitly setting a bin in the database.
- Added unit tests for the same, both reactive and normal
- Tested with annotations, YAML config, code config.

* Rolled back accidental push of wrong connection port.

* Format

* Cleanup and format

* Updated field name to storeAsBin

Changed the config field name from `storeInPkOnly` to `storeAsBin`,
added unit test to ensure config validation works.

* Rolled back test port to port 3000

* Remove old "storeInPkOnly" terminology, add javadocs, cleanup

* Apply suggestions from code review

Co-authored-by: Eugene R. <[email protected]>

* Revert change, causing NPE on runTestViaConfig() test

---------

Co-authored-by: roimenashe <[email protected]>
Co-authored-by: Roi Menashe <[email protected]>
Co-authored-by: Eugene R. <[email protected]>

Author: 3 people

README.md

@@ -484,7 +484,57 @@ public String getKey() {
 
 Note that it is not required to have a key on an object annotated with @AerospikeRecord. This is because an object can be embedded in another object (as a map or list) and hence not require a key to identify it to the database.
 
-Also, the existence of @AerospikeKey on a field does not imply that the field will get stored in the database explicitly. Use @AerospikeBin or mapAll attribute to ensure that the key gets mapped to the database too.
+Also, the existence of `@AerospikeKey` on a field does not imply that the field will get stored in the database explicitly. Use `@AerospikeBin` or `mapAll` attribute to ensure that the key gets mapped to the database too.
+
+By default, the key will always be stored in a separate column in the database. So for a class defined as
+
+```java
+@AerospikeRecord(namespace = "test", set = "testSet")
+public static class A {
+    @AerospikeKey
+    private long key;
+    private String value;
+}
+```
+
+there will be a bin in the database called `key`, whose value will be the same as the value used in the primary key. This is because Aerospike does not implicitly store the value of the key in the database, but rather uses a hash of the primary key as a unique representation. So the value in the database might look like:
+
+```
+aql> select * from test.testSet
++-----+--------+
+| key | value  |
++-----+--------+
+| 1   | "test" |
++-----+--------+
+```
+
+If it is desired to force the primary key to be stored in the database and NOT have key added explicitly as a column then two things must be set:
+
+1. The `@AerospikeRecord` annotation must have `sendKey = true`
+2. The `@AerospikeKey` annotation must have `storeAsBin = false`
+
+So the object would look like:
+
+```java
+@AerospikeRecord(namespace = "test", set = "testSet", sendKey = true)
+public static class A {
+    @AerospikeKey(storeAsBin = false)
+    private long key;
+    private String value;
+}
+```
+
+When data is inserted, the field `key` is not saved, but rather the key is saved as the primary key. When the value is read from the database, the stored primary key is put back into the `key` field. So the data in the database might be:
+
+```
+aql> select * from test.testSet
++----+--------+
+| PK | value  |
++----+--------+
+| 1  | "test" |
++----+--------+
+```
+
 
 ----
 
@@ -679,7 +729,7 @@ Here are how standard Java types are mapped to Aerospike types:
 | Map<?,?> | Map |
 | Object Reference (@AerospikeRecord) | List or Map |
 
-These types are built into the converter. However, if you wish to change them, you can use a [Custom Object Converter](#Custom-Object-Converters). For example, if you want Dates stored in the database as a string, you could do:
+These types are built into the converter. However, if you wish to change them, you can use a [Custom Object Converter](#custom-object-converters). For example, if you want Dates stored in the database as a string, you could do:
 
 ```java
 public static class DateConverter {
@@ -1975,6 +2025,7 @@ The key structure is used to specify the key to a record. Keys are optional in s
 
 The key structure contains:
 - **field**: The name of the field which to which this key is mapped. If this is provided, the getter and setter cannot be provided.
+- **storeAsBin**: Store the primary key as a bin in the database, alternatively it is recommended to use the `sendKey` facility related to Aerospike to save the key in the record's metadata (and set this flag to false). When the record is read, the value will be pulled back and placed in the key field.
 - **getter**: The getter method used to populate the key. This must be used in conjunction with a setter method, and excludes the use of the field attribute.
 - **setter**: The setter method used to map data back to the Java key. This is used in conjunction with the getter method and precludes the use of the field attribute. Note that the return type of the getter must match the type of the first parameter of the setter, and the setter can have either 1 or 2 parameters, with the second (optional) parameter being either of type [com.aerospike.client.Key](https://www.aerospike.com/apidocs/java/com/aerospike/client/Key.html) or Object.
 

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

@@ -12,4 +12,9 @@
      * The setter attribute is used only on Methods where the method is used to set the key on lazy object instantiation
      */
     boolean setter() default false;
+
+    /**
+     * Store the key as an Aerospike Bin, alternatively you can use @AerospikeRecord.sendKey to store the key in the record's metadata
+     */
+    boolean storeAsBin() default true;
 }

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

@@ -221,7 +221,7 @@ private <T> T read(Policy readPolicy, @NotNull Class<T> clazz, @NotNull Key key,
             try {
                 ThreadLocalKeySaver.save(key);
                 LoadedObjectResolver.begin();
-                return mappingConverter.convertToObject(clazz, record, entry, resolveDependencies);
+                return mappingConverter.convertToObject(clazz, key, record, entry, resolveDependencies);
             } catch (ReflectiveOperationException e) {
                 throw new AerospikeException(e);
             } finally {
@@ -252,7 +252,7 @@ private <T> T[] readBatch(BatchPolicy batchPolicy, @NotNull Class<T> clazz, @Not
             } else {
                 try {
                     ThreadLocalKeySaver.save(keys[i]);
-                    T result = mappingConverter.convertToObject(clazz, records[i], entry, false);
+                    T result = mappingConverter.convertToObject(clazz, keys[i], records[i], entry, false);
                     results[i] = result;
                 } catch (ReflectiveOperationException e) {
                     throw new AerospikeException(e);
@@ -372,7 +372,7 @@ public <T> void scan(ScanPolicy policy, @NotNull Class<T> clazz, @NotNull Proces
         AtomicBoolean userTerminated = new AtomicBoolean(false);
         try {
             mClient.scanAll(policy, namespace, setName, (key, record) -> {
-                T object = this.getMappingConverter().convertToObject(clazz, record);
+                T object = this.getMappingConverter().convertToObject(clazz, key, record);
                 if (!processor.process(object)) {
                     userTerminated.set(true);
                     throw new AerospikeException.ScanTerminated();
@@ -420,7 +420,7 @@ public <T> void query(QueryPolicy policy, @NotNull Class<T> clazz, @NotNull Proc
         RecordSet recordSet = mClient.query(policy, statement);
         try {
             while (recordSet.next()) {
-                T object = this.getMappingConverter().convertToObject(clazz, recordSet.getRecord());
+                T object = this.getMappingConverter().convertToObject(clazz, recordSet.getKey(), recordSet.getRecord());
                 if (!processor.process(object)) {
                     break;
                 }

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

@@ -64,6 +64,7 @@ public class ClassCacheEntry<T> {
     private final Class<T> clazz;
     private ValueType key;
     private String keyName = null;
+    private boolean keyAsBin = true;
     private final TreeMap<String, ValueType> values = new TreeMap<>();
     private ClassCacheEntry<?> superClazz;
     private int binCount;
@@ -416,12 +417,12 @@ private Method findConstructorFactoryMethod() {
         if (!StringUtils.isBlank(this.factoryClass) || !StringUtils.isBlank(this.factoryMethod)) {
             // Both must be specified
             if (StringUtils.isBlank(this.factoryClass)) {
-                throw new AerospikeException("Missing factoryClass definition when factoryMethod is specified on class " +
-                        clazz.getSimpleName());
+                throw new AerospikeException(String.format("Missing factoryClass definition when factoryMethod is specified on class %s",
+                        clazz.getSimpleName()));
             }
             if (StringUtils.isBlank(this.factoryClass)) {
-                throw new AerospikeException("Missing factoryMethod definition when factoryClass is specified on class " +
-                        clazz.getSimpleName());
+                throw new AerospikeException(String.format("Missing factoryMethod definition when factoryClass is specified on class %s",
+                        clazz.getSimpleName()));
             }
             // Load the class and check for the method
             try {
@@ -479,8 +480,8 @@ private void setConstructorFactoryMethod(Method method) {
     private void findConstructor() {
         Constructor<?>[] constructors = clazz.getDeclaredConstructors();
         if (constructors.length == 0) {
-            throw new AerospikeException("Class " + clazz.getSimpleName() +
-                    " has no constructors and hence cannot be mapped to Aerospike");
+            throw new AerospikeException(String.format("Class %s has no constructors and hence cannot be mapped to Aerospike",
+                    clazz.getSimpleName()));
         }
         Constructor<?> desiredConstructor = null;
         Constructor<?> noArgConstructor = null;
@@ -494,9 +495,9 @@ private void findConstructor() {
                 AerospikeConstructor aerospikeConstructor = thisConstructor.getAnnotation(AerospikeConstructor.class);
                 if (aerospikeConstructor != null) {
                     if (desiredConstructor != null) {
-                        throw new AerospikeException("Class " + clazz.getSimpleName() +
-                                " has multiple constructors annotated with @AerospikeConstructor. " +
-                                "Only one constructor can be so annotated.");
+                        throw new AerospikeException(String.format("Class %s" +
+                                " has multiple constructors annotated with @AerospikeConstructor." +
+                                " Only one constructor can be so annotated.", clazz.getSimpleName()));
                     } else {
                         desiredConstructor = thisConstructor;
                     }
@@ -509,8 +510,9 @@ private void findConstructor() {
         }
 
         if (desiredConstructor == null) {
-            throw new AerospikeException("Class " + clazz.getSimpleName() + " has neither a no-arg constructor, " +
-                    "nor a constructor annotated with @AerospikeConstructor so cannot be mapped to Aerospike.");
+            throw new AerospikeException(String.format("Class %s has neither a no-arg constructor, " +
+                    "nor a constructor annotated with @AerospikeConstructor so cannot be mapped to Aerospike.",
+                    clazz.getSimpleName()));
         }
 
         Parameter[] params = desiredConstructor.getParameters();
@@ -551,10 +553,11 @@ private void findConstructor() {
             }
             Class<?> type = thisParam.getType();
             if (!type.isAssignableFrom(allValues.get(binName).getType())) {
-                throw new AerospikeException("Class " + clazz.getSimpleName() + " has a preferred constructor of " +
-                        desiredConstructor + ". However, parameter " + count +
-                        " is of type " + type + " but assigned from bin \"" + binName + "\" of type " +
-                        values.get(binName).getType() + ". These types are incompatible.");
+                throw new AerospikeException(String.format("Class %s has a preferred constructor of" +
+                                " %s. However, parameter %s" +
+                                " is of type %s but assigned from bin \"%s\" of type %s." +
+                                " These types are incompatible.",
+                        clazz.getSimpleName(), desiredConstructor, count, type, binName, values.get(binName).getType()));
             }
             constructorParamBins[count - 1] = binName;
             constructorParamDefaults[count - 1] = PrimitiveDefaults.getDefaultValue(thisParam.getType());
@@ -627,20 +630,24 @@ private void loadPropertiesFromClass() {
 
         if (keyProperty != null) {
             keyProperty.validate(clazz.getName(), config, true);
+
             if (key != null) {
-                throw new AerospikeException("Class " + clazz.getName() + " cannot have a more than one key");
+                throw new AerospikeException(String.format("Class %s cannot have more than one key", clazz.getName()));
             }
+
             AnnotatedType annotatedType = new AnnotatedType(config, keyProperty.getGetter());
             TypeMapper typeMapper = TypeUtils.getMapper(keyProperty.getType(), annotatedType, this.mapper);
             this.key = new ValueType.MethodValue(keyProperty, typeMapper, annotatedType);
         }
         for (String thisPropertyName : properties.keySet()) {
             PropertyDefinition thisProperty = properties.get(thisPropertyName);
             thisProperty.validate(clazz.getName(), config, false);
+
             if (this.values.get(thisPropertyName) != null) {
-                throw new AerospikeException("Class " + clazz.getName() + " cannot define the mapped name " +
-                        thisPropertyName + " more than once");
+                throw new AerospikeException(String.format("Class %s cannot define the mapped name %s more than once",
+                        clazz.getName(), thisPropertyName));
             }
+
             AnnotatedType annotatedType = new AnnotatedType(config, thisProperty.getGetter());
             TypeMapper typeMapper = TypeUtils.getMapper(thisProperty.getType(), annotatedType, this.mapper);
             ValueType value = new ValueType.MethodValue(thisProperty, typeMapper, annotatedType);
@@ -654,20 +661,38 @@ private void loadFieldsFromClass() {
         for (Field thisField : this.clazz.getDeclaredFields()) {
             boolean isKey = false;
             BinConfig thisBin = getBinFromField(thisField);
+
             if (Modifier.isFinal(thisField.getModifiers()) && Modifier.isStatic(thisField.getModifiers())) {
             	// We cannot map static final fields
             	continue;
             }
+
             if (thisField.isAnnotationPresent(AerospikeKey.class) || (!StringUtils.isBlank(keyField) && keyField.equals(thisField.getName()))) {
                 if (thisField.isAnnotationPresent(AerospikeExclude.class) || (thisBin != null && thisBin.isExclude() != null && thisBin.isExclude())) {
-                    throw new AerospikeException("Class " + clazz.getName() + " cannot have a field which is both a key and excluded.");
+                    throw new AerospikeException(String.format("Class %s cannot have a field which is both a key and excluded.",
+                            clazz.getName()));
                 }
+
                 if (key != null) {
-                    throw new AerospikeException("Class " + clazz.getName() + " cannot have a more than one key");
+                    throw new AerospikeException(String.format("Class %s cannot have more than one key",
+                            clazz.getName()));
+                }
+                AerospikeKey keyAnnotation = thisField.getAnnotation(AerospikeKey.class);
+                boolean storeAsBin = keyAnnotation == null || keyAnnotation.storeAsBin();
+
+                if (keyConfig != null && keyConfig.getStoreAsBin() != null) {
+                    storeAsBin = keyConfig.getStoreAsBin();
                 }
+
+                if (!storeAsBin && (this.sendKey == null || !this.sendKey)) {
+                    throw new AerospikeException(String.format("Class %s attempts to store primary key information" +
+                            " inside the aerospike key, but sendKey is not true at the record level", clazz.getName()));
+                }
+
                 AnnotatedType annotatedType = new AnnotatedType(config, thisField);
                 TypeMapper typeMapper = TypeUtils.getMapper(thisField.getType(), annotatedType, this.mapper);
                 this.key = new ValueType.FieldValue(thisField, typeMapper, annotatedType);
+                this.keyAsBin = storeAsBin;
                 isKey = true;
             }
 
@@ -694,7 +719,8 @@ private void loadFieldsFromClass() {
                 }
 
                 if (this.values.get(name) != null) {
-                    throw new AerospikeException("Class " + clazz.getName() + " cannot define the mapped name " + name + " more than once");
+                    throw new AerospikeException(String.format("Class %s cannot define the mapped name %s more than once",
+                            clazz.getName(), name));
                 }
                 if ((bin != null && bin.useAccessors()) || (thisBin != null && thisBin.getUseAccessors() != null && thisBin.getUseAccessors())) {
                     validateAccessorsForField(name, thisField);
@@ -778,8 +804,8 @@ public Object getKey(Object object) {
         try {
             Object key = this._getKey(object);
             if (key == null) {
-                throw new AerospikeException("Null key from annotated object of class " + this.clazz.getSimpleName() +
-                        ". Did you forget an @AerospikeKey annotation?");
+                throw new AerospikeException(String.format("Null key from annotated object of class %s." +
+                        " Did you forget an @AerospikeKey annotation?", this.clazz.getSimpleName()));
             }
             return key;
         } catch (ReflectiveOperationException re) {
@@ -850,6 +876,10 @@ public Bin[] getBins(Object instance, boolean allowNullBins, String[] binNames)
             while (thisClass != null) {
                 Set<String> keys = thisClass.values.keySet();
                 for (String name : keys) {
+                    if (name.equals(thisClass.keyName) && !thisClass.keyAsBin) {
+                        // Do not explicitly write the key to the bin
+                        continue;
+                    }
                     if (contains(binNames, name)) {
                         ValueType value = (ValueType) thisClass.values.get(name);
                         Object javaValue = value.get(instance);
@@ -948,15 +978,15 @@ public List<Object> getList(Object instance, boolean skipKey, boolean needsType)
     }
 
     public T constructAndHydrate(Map<String, Object> map) {
-        return constructAndHydrate(null, map);
+        return constructAndHydrate(null, null, map);
     }
 
-    public T constructAndHydrate(Record record) {
-        return constructAndHydrate(record, null);
+    public T constructAndHydrate(Key key, Record record) {
+        return constructAndHydrate(key, record, null);
     }
 
     @SuppressWarnings("unchecked")
-    private T constructAndHydrate(Record record, Map<String, Object> map) {
+    private T constructAndHydrate(Key key, Record record, Map<String, Object> map) {
         Map<String, Object> valueMap = new HashMap<>();
         try {
             ClassCacheEntry<?> thisClass = this;
@@ -976,7 +1006,19 @@ private T constructAndHydrate(Record record, Map<String, Object> map) {
             while (thisClass != null) {
                 for (String name : thisClass.values.keySet()) {
                     ValueType value = thisClass.values.get(name);
-                    Object aerospikeValue = record == null ? map.get(name) : record.getValue(name);
+                    Object aerospikeValue;
+                    if (record == null) {
+                        aerospikeValue = map.get(name);
+                    } else if (name.equals(thisClass.keyName) && !thisClass.keyAsBin) {
+                        if (key.userKey != null) {
+                            aerospikeValue = key.userKey.getObject();
+                        } else {
+                            throw new AerospikeException(String.format("Key field on class %s was <null> for key %s." +
+                                    " Was the record saved passing 'sendKey = true'? ", className, key));
+                        }
+                    } else {
+                        aerospikeValue = record.getValue(name);
+                    }
                     valueMap.put(name, value.getTypeMapper().fromAerospikeFormat(aerospikeValue));
                 }
                 if (result == null) {