DDBEnhanced - Support to flatten a Map into top level attributes of the object

.changes/next-release/feature-AmazonDynamoDBEnhancedClient-22723fc.json

@@ -0,0 +1,6 @@
+{
+    "type": "feature",
+    "category": "Amazon DynamoDB Enhanced Client",
+    "contributor": "",
+    "description": "Added the support to flatten a Map into top level attributes of the object"
+}

services-custom/dynamodb-enhanced/README.md

@@ -682,3 +682,53 @@ private static final StaticTableSchema<Customer> CUSTOMER_TABLE_SCHEMA =
 ```
 Just as for annotations, you can flatten as many different eligible classes as you like using the
 builder pattern. 
+
+#### Using composition
+
+Using composition, the @DynamoDBFlattenMap annotation support to flatten a Map:
+```java
+@DynamoDbBean
+public class Customer {
+    private String name;
+    private String city;
+    private String address;
+
+    private Map<String, String> detailsMap;
+
+    public String getName() { return this.name; }
+    public void setName(String name) { this.name = name;}
+    public String getCity() { return this.city; }
+    public void setCity(String city) { this.city = city;}
+    public String getAddress() { return this.address; }
+    public void setAddress(String address) { this.name = address;}
+
+    @DynamoDbFlattenMap
+    public Map<String, String> getDetailsMap() { return this.detailsMap; }
+    public void setDetailsMap(Map<String, String> record) { this.detailsMap = detailsMap;}
+}
+```
+You can flatten only one map present on a record, otherwise it will be thrown an exception
+
+Flat map composite classes using StaticTableSchema:
+
+```java
+@Data
+public class Customer {
+  private String name;
+  private String city;
+  private String address;
+  private Map<String, String> detailsMap;
+  //getters and setters for all attributes
+}
+
+private static final StaticTableSchema<Customer> CUSTOMER_TABLE_SCHEMA =
+  StaticTableSchema.builder(Customer.class)
+    .newItemSupplier(Customer::new)
+    .addAttribute(String.class, a -> a.name("name")
+                                      .getter(Customer::getName)
+                                      .setter(Customer::setName))
+    // Because we are flattening a Map object, we supply a getter and setter so the
+    // mapper knows how to access it
+    .flattenMap(Map::detailsMap, Map::detailsMap)
+    .build(); 
+```

services-custom/dynamodb-enhanced/src/it/java/software/amazon/awssdk/enhanced/dynamodb/DynamoDbEnhancedIntegrationTestBase.java

@@ -21,6 +21,7 @@
 import static software.amazon.awssdk.enhanced.dynamodb.mapper.StaticAttributeTags.secondarySortKey;
 
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
 import java.util.UUID;
 import java.util.stream.Collectors;
@@ -75,6 +76,36 @@ protected static DynamoDbAsyncClient createAsyncDynamoDbClient() {
                                                            .setter(Record::setStringAttribute))
                          .build();
 
+    protected static final TableSchema<Record> RECORD_WITH_FLATTEN_MAP_TABLE_SCHEMA =
+        StaticTableSchema.builder(Record.class)
+                         .newItemSupplier(Record::new)
+                         .addAttribute(String.class, a -> a.name("id")
+                                                           .getter(Record::getId)
+                                                           .setter(Record::setId)
+                                                           .tags(primaryPartitionKey(), secondaryPartitionKey("index1")))
+                         .addAttribute(Integer.class, a -> a.name("sort")
+                                                            .getter(Record::getSort)
+                                                            .setter(Record::setSort)
+                                                            .tags(primarySortKey(), secondarySortKey("index1")))
+                         .addAttribute(Integer.class, a -> a.name("value")
+                                                            .getter(Record::getValue)
+                                                            .setter(Record::setValue))
+                         .addAttribute(String.class, a -> a.name("gsi_id")
+                                                           .getter(Record::getGsiId)
+                                                           .setter(Record::setGsiId)
+                                                           .tags(secondaryPartitionKey("gsi_keys_only")))
+                         .addAttribute(Integer.class, a -> a.name("gsi_sort")
+                                                            .getter(Record::getGsiSort)
+                                                            .setter(Record::setGsiSort)
+                                                            .tags(secondarySortKey("gsi_keys_only")))
+                         .addAttribute(String.class, a -> a.name("stringAttribute")
+                                                           .getter(Record::getStringAttribute)
+                                                           .setter(Record::setStringAttribute))
+                         .flatten("attributesMap",
+                                  Record::getAttributesMap,
+                                  Record::setAttributesMap)
+                         .build();
+
 
     protected static final List<Record> RECORDS =
         IntStream.range(0, 9)
@@ -87,6 +118,22 @@ protected static DynamoDbAsyncClient createAsyncDynamoDbClient() {
                      .setGsiSort(i))
                  .collect(Collectors.toList());
 
+    protected static final List<Record> RECORDS_WITH_FLATTEN_MAP =
+        IntStream.range(0, 9)
+                 .mapToObj(i -> new Record()
+                     .setId("id-value")
+                     .setSort(i)
+                     .setValue(i)
+                     .setStringAttribute(getStringAttrValue(10 * 1024))
+                     .setGsiId("gsi-id-value")
+                     .setGsiSort(i)
+                     .setAttributesMap(new HashMap<String, String>() {{
+                         put("mapAttribute1", "mapValue1");
+                         put("mapAttribute2", "mapValue2");
+                         put("mapAttribute3", "mapValue3");
+                     }}))
+                 .collect(Collectors.toList());
+
     protected static final List<Record> KEYS_ONLY_RECORDS =
         RECORDS.stream()
                .map(record -> new Record()

services-custom/dynamodb-enhanced/src/it/java/software/amazon/awssdk/enhanced/dynamodb/ScanQueryWithFlattenMapIntegrationTest.java

@@ -0,0 +1,117 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.enhanced.dynamodb;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
+import static software.amazon.awssdk.enhanced.dynamodb.internal.AttributeValues.numberValue;
+import static software.amazon.awssdk.enhanced.dynamodb.internal.AttributeValues.stringValue;
+import static software.amazon.awssdk.enhanced.dynamodb.model.QueryConditional.sortBetween;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import software.amazon.awssdk.enhanced.dynamodb.model.Page;
+import software.amazon.awssdk.enhanced.dynamodb.model.QueryEnhancedRequest;
+import software.amazon.awssdk.enhanced.dynamodb.model.Record;
+import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+
+public class ScanQueryWithFlattenMapIntegrationTest extends DynamoDbEnhancedIntegrationTestBase {
+
+    private static final String TABLE_NAME = createTestTableName();
+
+    private static DynamoDbClient dynamoDbClient;
+    private static DynamoDbEnhancedClient enhancedClient;
+    private static DynamoDbTable<Record> mappedTable;
+
+    @BeforeClass
+    public static void setup() {
+        dynamoDbClient = createDynamoDbClient();
+        enhancedClient = DynamoDbEnhancedClient.builder().dynamoDbClient(dynamoDbClient).build();
+        mappedTable = enhancedClient.table(TABLE_NAME, RECORD_WITH_FLATTEN_MAP_TABLE_SCHEMA);
+        mappedTable.createTable();
+        dynamoDbClient.waiter().waitUntilTableExists(r -> r.tableName(TABLE_NAME));
+    }
+
+    @AfterClass
+    public static void teardown() {
+        try {
+            dynamoDbClient.deleteTable(r -> r.tableName(TABLE_NAME));
+        } finally {
+            dynamoDbClient.close();
+        }
+    }
+
+    private void insertRecords() {
+        RECORDS_WITH_FLATTEN_MAP.forEach(record -> mappedTable.putItem(r -> r.item(record)));
+    }
+
+    @Test
+    public void queryWithFlattenMapRecord_correctlyRetrievesProjectedAttributes() {
+        insertRecords();
+
+        Iterator<Page<Record>> results =
+            mappedTable.query(QueryEnhancedRequest.builder()
+                                                  .queryConditional(sortBetween(k-> k.partitionValue("id-value").sortValue(2),
+                                                                                k-> k.partitionValue("id-value").sortValue(6)))
+                                                  .attributesToProject("mapAttribute1", "mapAttribute2")
+                                                  .limit(3)
+                                                  .build())
+                       .iterator();
+
+        Page<Record> page1 = results.next();
+        assertThat(results.hasNext(), is(true));
+        Page<Record> page2 = results.next();
+        assertThat(results.hasNext(), is(false));
+
+        Map<String, String> resultedAttributesMap = new HashMap<>();
+        resultedAttributesMap.put("mapAttribute1", "mapValue1");
+        resultedAttributesMap.put("mapAttribute2", "mapValue2");
+
+        List<Record> page1Items = page1.items();
+        assertThat(page1Items.size(), is(3));
+        assertThat(page1Items.get(0).getAttributesMap(), is(resultedAttributesMap));
+        assertThat(page1Items.get(1).getAttributesMap(), is(resultedAttributesMap));
+        assertThat(page1Items.get(2).getAttributesMap(), is(resultedAttributesMap));
+        assertThat(page1.consumedCapacity(), is(nullValue()));
+        assertThat(page1.lastEvaluatedKey(), is(getKeyMap(4)));
+        assertThat(page1.count(), equalTo(3));
+        assertThat(page1.scannedCount(), equalTo(3));
+
+        List<Record> page2Items = page2.items();
+        assertThat(page2Items.size(), is(2));
+        assertThat(page2Items.get(0).getAttributesMap(), is(resultedAttributesMap));
+        assertThat(page2Items.get(1).getAttributesMap(), is(resultedAttributesMap));
+        assertThat(page2.lastEvaluatedKey(), is(nullValue()));
+        assertThat(page2.count(), equalTo(2));
+        assertThat(page2.scannedCount(), equalTo(2));
+    }
+
+    private Map<String, AttributeValue> getKeyMap(int sort) {
+        Map<String, AttributeValue> result = new HashMap<>();
+        result.put("id", stringValue(RECORDS.get(sort).getId()));
+        result.put("sort", numberValue(RECORDS.get(sort).getSort()));
+        return Collections.unmodifiableMap(result);
+    }
+}

services-custom/dynamodb-enhanced/src/it/java/software/amazon/awssdk/enhanced/dynamodb/model/Record.java

@@ -15,6 +15,7 @@
 
 package software.amazon.awssdk.enhanced.dynamodb.model;
 
+import java.util.Map;
 import java.util.Objects;
 
 public class Record {
@@ -27,6 +28,8 @@ public class Record {
 
     private String stringAttribute;
 
+    private Map<String, String> attributesMap;
+
     public String getId() {
         return id;
     }
@@ -81,6 +84,15 @@ public Record setStringAttribute(String stringAttribute) {
         return this;
     }
 
+    public Map<String, String> getAttributesMap() {
+        return attributesMap;
+    }
+
+    public Record setAttributesMap(Map<String, String> attributesMap) {
+        this.attributesMap = attributesMap;
+        return this;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;
@@ -91,11 +103,12 @@ public boolean equals(Object o) {
                Objects.equals(value, record.value) &&
                Objects.equals(gsiId, record.gsiId) &&
                Objects.equals(stringAttribute, record.stringAttribute) &&
-               Objects.equals(gsiSort, record.gsiSort);
+               Objects.equals(gsiSort, record.gsiSort) &&
+               Objects.equals(attributesMap, record.attributesMap);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(id, sort, value, gsiId, gsiSort, stringAttribute);
+        return Objects.hash(id, sort, value, gsiId, gsiSort, stringAttribute, attributesMap);
     }
 }

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

@@ -35,6 +35,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.WeakHashMap;
 import java.util.function.BiConsumer;
@@ -63,6 +64,7 @@
 import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbBean;
 import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbConvertedBy;
 import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbFlatten;
+import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbFlattenMap;
 import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbIgnore;
 import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbIgnoreNulls;
 import software.amazon.awssdk.enhanced.dynamodb.mapper.annotations.DynamoDbImmutable;
@@ -222,6 +224,14 @@ private static <T> StaticTableSchema<T> createStaticTableSchema(Class<T> beanCla
             throw new IllegalArgumentException(e);
         }
 
+        List<PropertyDescriptor> mappableProperties = Arrays.stream(beanInfo.getPropertyDescriptors())
+                                                            .filter(p -> isMappableProperty(beanClass, p))
+                                                            .collect(Collectors.toList());
+
+        if (dynamoDbFlattenMapAnnotationHasInvalidUse(mappableProperties)) {
+            throw new IllegalArgumentException("More than one @DynamoDbFlattenMap annotation found on the same record");
+        }
+
         Supplier<T> newObjectSupplier = newObjectSupplierForClass(beanClass, lookup);
 
         StaticTableSchema.Builder<T> builder = StaticTableSchema.builder(beanClass)
@@ -231,29 +241,35 @@ private static <T> StaticTableSchema<T> createStaticTableSchema(Class<T> beanCla
 
         List<StaticAttribute<T, ?>> attributes = new ArrayList<>();
 
-        Arrays.stream(beanInfo.getPropertyDescriptors())
-              .filter(p -> isMappableProperty(beanClass, p))
-              .forEach(propertyDescriptor -> {
+        mappableProperties.forEach(propertyDescriptor -> {
                   DynamoDbFlatten dynamoDbFlatten = getPropertyAnnotation(propertyDescriptor, DynamoDbFlatten.class);
 
                   if (dynamoDbFlatten != null) {
                       builder.flatten(TableSchema.fromClass(propertyDescriptor.getReadMethod().getReturnType()),
                                       getterForProperty(propertyDescriptor, beanClass, lookup),
                                       setterForProperty(propertyDescriptor, beanClass, lookup));
                   } else {
-                      AttributeConfiguration attributeConfiguration =
-                          resolveAttributeConfiguration(propertyDescriptor);
+                      DynamoDbFlattenMap dynamoDbFlattenMap = getPropertyAnnotation(propertyDescriptor, DynamoDbFlattenMap.class);
+
+                      if (dynamoDbFlattenMap != null) {
+                          builder.flatten(propertyDescriptor.getName(), getterForProperty(propertyDescriptor, beanClass, lookup),
+                                          setterForProperty(propertyDescriptor, beanClass, lookup));
 
-                      StaticAttribute.Builder<T, ?> attributeBuilder =
-                          staticAttributeBuilder(propertyDescriptor, beanClass, lookup, metaTableSchemaCache,
-                                                 attributeConfiguration);
+                      } else {
+                          AttributeConfiguration attributeConfiguration =
+                              resolveAttributeConfiguration(propertyDescriptor);
 
-                      Optional<AttributeConverter> attributeConverter =
+                          StaticAttribute.Builder<T, ?> attributeBuilder =
+                              staticAttributeBuilder(propertyDescriptor, beanClass, lookup, metaTableSchemaCache,
+                                                     attributeConfiguration);
+
+                          Optional<AttributeConverter> attributeConverter =
                               createAttributeConverterFromAnnotation(propertyDescriptor, lookup);
-                      attributeConverter.ifPresent(attributeBuilder::attributeConverter);
+                          attributeConverter.ifPresent(attributeBuilder::attributeConverter);
 
-                      addTagsToAttribute(attributeBuilder, propertyDescriptor);
-                      attributes.add(attributeBuilder.build());
+                          addTagsToAttribute(attributeBuilder, propertyDescriptor);
+                          attributes.add(attributeBuilder.build());
+                      }
                   }
               });
 
@@ -286,6 +302,15 @@ private static Optional<Method> findFluentSetter(Class<?> beanClass, String prop
                      .findFirst();
     }
 
+    private static boolean dynamoDbFlattenMapAnnotationHasInvalidUse(List<PropertyDescriptor> mappableProperties) {
+        return mappableProperties.stream()
+                                 .map(pd -> getPropertyAnnotation(pd, DynamoDbFlattenMap.class))
+                                 .filter(Objects::nonNull)
+                                 .skip(1)
+                                 .findFirst()
+                                 .isPresent();
+    }
+
     private static AttributeConfiguration resolveAttributeConfiguration(PropertyDescriptor propertyDescriptor) {
         boolean shouldPreserveEmptyObject = getPropertyAnnotation(propertyDescriptor,
                                                                   DynamoDbPreserveEmptyObject.class) != null;