Add support for starting DynamoDB Record Version with explicit value

The prior behavior required that a version be initialized with a null value,
this required mapper clients to use Integer instead of the int primitive.
This change allows clients to explicitly initialize the version to a value
which makes it simpler for clients to use primitive values and potentially
avoid null pointer exceptions and checks.

The default starting value of 0 and increment value of 1 are intended to
provide sane defaults that are identical to the existing behavior while
enabling clients to have more fine-graned control over how the versioning
is managed for their specific use-cases.

The current implementation configures the values at the extension level only
but the implementation can be expanded to gather the value from the model
annotation to customize the values on a per table basis.

.changes/next-release/feature-AmazonDyanmoDBEnhancedClient-2047f83.json

@@ -0,0 +1,6 @@
+{
+    "category": "Amazon DyanmoDB Enhanced Client", 
+    "contributor": "kiesler", 
+    "type": "feature", 
+    "description": "DynamoDB Enhanced Client Versioned Record can start at 0"
+}

services-custom/dynamodb-enhanced/src/main/java/software/amazon/awssdk/enhanced/dynamodb/Expression.java

@@ -311,6 +311,15 @@ public int hashCode() {
         return result;
     }
 
+    @Override
+    public String toString() {
+        return "Expression{" +
+               "expression='" + expression + '\'' +
+               ", expressionValues=" + expressionValues +
+               ", expressionNames=" + expressionNames +
+               '}';
+    }
+
     /**
      * A builder for {@link Expression}
      */

services-custom/dynamodb-enhanced/src/main/java/software/amazon/awssdk/enhanced/dynamodb/extensions/VersionedRecordExtension.java

@@ -61,8 +61,20 @@ public final class VersionedRecordExtension implements DynamoDbEnhancedClientExt
     private static final Function<String, String> VERSIONED_RECORD_EXPRESSION_VALUE_KEY_MAPPER = key -> ":old_" + key + "_value";
     private static final String CUSTOM_METADATA_KEY = "VersionedRecordExtension:VersionAttribute";
     private static final VersionAttribute VERSION_ATTRIBUTE = new VersionAttribute();
-
-    private VersionedRecordExtension() {
+    private static final AttributeValue DEFAULT_VALUE = AttributeValue.fromNul(Boolean.TRUE);
+
+    private final int startingValue;
+    private final int increment;
+
+    /**
+     * Creates a new {@link VersionedRecordExtension} using the supplied starting and incrementing value.
+     *
+     * @param startingValue the value used to compare if a record is the initial version of a record.
+     * @param increment the amount to increment the version by with each subsequent update.
+     */
+    private VersionedRecordExtension(int startingValue, int increment) {
+        this.startingValue = startingValue;
+        this.increment = increment;
     }
 
     public static Builder builder() {
@@ -119,23 +131,24 @@ public WriteModification beforeWrite(DynamoDbExtensionContext.BeforeWrite contex
 
     private Pair<AttributeValue, Expression> getRecordUpdates(String versionAttributeKey,
                                                               Map<String, AttributeValue> itemToTransform) {
-        Optional<AttributeValue> existingVersionValue =
-            Optional.ofNullable(itemToTransform.get(versionAttributeKey));
+        // Default to NUL if not present to reduce additional checks further along
+        AttributeValue existingVersionValue = itemToTransform.getOrDefault(versionAttributeKey, DEFAULT_VALUE);
 
         if (isInitialVersion(existingVersionValue)) {
             // First version of the record ensure it does not exist
             return createInitialRecord(versionAttributeKey);
         }
         // Existing record, increment version
-        return updateExistingRecord(versionAttributeKey, existingVersionValue.get());
+        return updateExistingRecord(versionAttributeKey, existingVersionValue);
     }
 
-    private boolean isInitialVersion(Optional<AttributeValue> existingVersionValue) {
-        return !existingVersionValue.isPresent() || isNullAttributeValue(existingVersionValue.get());
+    private boolean isInitialVersion(AttributeValue existingVersionValue) {
+        return isNullAttributeValue(existingVersionValue)
+               || getExistingVersion(existingVersionValue) == this.startingValue;
     }
 
     private Pair<AttributeValue, Expression> createInitialRecord(String versionAttributeKey) {
-        AttributeValue newVersionValue = incrementVersion(0);
+        AttributeValue newVersionValue = incrementVersion(this.startingValue);
 
         String attributeKeyRef = keyRef(versionAttributeKey);
 
@@ -177,16 +190,43 @@ private int getExistingVersion(AttributeValue existingVersionValue) {
     }
 
     private AttributeValue incrementVersion(int version) {
-        return AttributeValue.fromN(Integer.toString(version + 1));
+        return AttributeValue.fromN(Integer.toString(version + this.increment));
     }
 
     @NotThreadSafe
     public static final class Builder {
+        private int startingValue = 0;
+        private int increment = 1;
+
         private Builder() {
         }
 
+        /**
+         * Sets the startingValue used to compare if a record is the initial version of a record.
+         * Default value - {@code 0}.
+         *
+         * @param startingValue
+         * @return the builder instance
+         */
+        public Builder startAt(int startingValue) {
+            this.startingValue = startingValue;
+            return this;
+        }
+
+        /**
+         * Sets the amount to increment the version by with each subsequent update.
+         * Default value - {@code 1}.
+         *
+         * @param increment
+         * @return the builder instance
+         */
+        public Builder incrementBy(int increment) {
+            this.increment = increment;
+            return this;
+        }
+
         public VersionedRecordExtension build() {
-            return new VersionedRecordExtension();
+            return new VersionedRecordExtension(this.startingValue, this.increment);
         }
     }
 }

services-custom/dynamodb-enhanced/src/test/java/software/amazon/awssdk/enhanced/dynamodb/extensions/VersionedRecordExtensionTest.java

@@ -112,6 +112,33 @@ public void beforeWrite_initialVersionDueToExplicitNull_transformedItemIsCorrect
         assertThat(result.transformedItem(), is(fakeItemWithInitialVersion));
     }
 
+    @Test
+    public void beforeWrite_initialVersionDueToExplicitZero_expressionAndTransformedItemIsCorrect() {
+        FakeItem fakeItem = createUniqueFakeItem();
+
+        Map<String, AttributeValue> inputMap =
+            new HashMap<>(FakeItem.getTableSchema().itemToMap(fakeItem, true));
+        inputMap.put("version", AttributeValue.builder().n("0").build());
+
+        Map<String, AttributeValue> fakeItemWithInitialVersion =
+            new HashMap<>(FakeItem.getTableSchema().itemToMap(fakeItem, true));
+        fakeItemWithInitialVersion.put("version", AttributeValue.builder().n("1").build());
+
+        WriteModification result =
+            versionedRecordExtension.beforeWrite(DefaultDynamoDbExtensionContext
+                                                     .builder()
+                                                     .items(inputMap)
+                                                     .tableMetadata(FakeItem.getTableMetadata())
+                                                     .operationContext(PRIMARY_CONTEXT).build());
+
+        assertThat(result.transformedItem(), is(fakeItemWithInitialVersion));
+        assertThat(result.additionalConditionalExpression(),
+                   is(Expression.builder()
+                                .expression("attribute_not_exists(#AMZN_MAPPED_version)")
+                                .expressionNames(singletonMap("#AMZN_MAPPED_version", "version"))
+                                .build()));
+    }
+
     @Test
     public void beforeWrite_existingVersion_expressionIsCorrect() {
         FakeItem fakeItem = createUniqueFakeItem();