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": "Add support for @DynamoDbFlatten to flatten a Map<String, String> to top level attributes of an object"
+}

services-custom/dynamodb-enhanced/README.md

@@ -613,7 +613,7 @@ private static final StaticTableSchema<Customer> CUSTOMER_TABLE_SCHEMA =
 ```
 #### Using composition
 
-Using composition, the @DynamoDbFlatten annotation flat maps the composite class:
+Using composition, the @DynamoDbFlatten annotation flat maps composite classes and Map attributes:
 ```java
 @DynamoDbBean
 public class Customer {
@@ -640,10 +640,40 @@ public class GenericRecord {
     public void setCreatedDate(String createdDate) { this.createdDate = createdDate;}
 }
 ```
-You can flatten as many different eligible classes as you like using the flatten annotation.
+
+You can also apply the @DynamoDbFlatten annotation to flatten a Map into top-level attributes:
+```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.address = address;}
+
+    @DynamoDbFlatten
+    public Map<String, String> getDetailsMap() { return this.detailsMap; }
+    public void setDetailsMap(Map<String, String> detailsMap) { this.detailsMap = detailsMap;}
+}
+```
+
+**Object Flattening**: You can flatten as many different eligible classes as you like using the flatten annotation.
 The only constraints are that attributes must not have the same name when they are being rolled
 together, and there must never be more than one partition key, sort key or table name.
 
+**Map Flattening**: 
+- A record can contain at most one `@DynamoDbFlatten` on a Map property. This limit applies across the entire class hierarchy, including any composed or flattened classes.
+- The flattened map must use `String` as both the key and value type (`Map<String, String>`). Other key or value types are not supported.
+- Attribute names generated from map keys must not conflict with existing attributes on the record. If a conflict is detected, an exception will be thrown.
+- If more than one flattened map is present, an exception will be thrown during schema creation.
+- Other annotations like `@DynamoDbUpdateBehavior` are not supported on flattened maps, consistent with the existing object flattening behavior.
+
 Flat map composite classes using StaticTableSchema:
 
 ```java
@@ -685,3 +715,26 @@ 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. 
+
+For map flattening 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(Customer::getDetailsMap, Customer::setDetailsMap)
+    .build(); 
+```

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

@@ -0,0 +1,170 @@
+/*
+ * 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.mapper.StaticAttributeTags.primaryPartitionKey;
+import static software.amazon.awssdk.enhanced.dynamodb.mapper.StaticAttributeTags.primarySortKey;
+import static software.amazon.awssdk.enhanced.dynamodb.mapper.StaticAttributeTags.secondaryPartitionKey;
+import static software.amazon.awssdk.enhanced.dynamodb.mapper.StaticAttributeTags.secondarySortKey;
+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 java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import software.amazon.awssdk.enhanced.dynamodb.mapper.StaticTableSchema;
+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 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();
+
+    private 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());
+
+    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> expectedAttributesMap = new HashMap<>();
+        expectedAttributesMap.put("mapAttribute1", "mapValue1");
+        expectedAttributesMap.put("mapAttribute2", "mapValue2");
+
+        List<Record> page1Items = page1.items();
+        assertThat(page1Items.size(), is(3));
+        assertThat(page1Items.get(0).getAttributesMap(), is(expectedAttributesMap));
+        assertThat(page1Items.get(1).getAttributesMap(), is(expectedAttributesMap));
+        assertThat(page1Items.get(2).getAttributesMap(), is(expectedAttributesMap));
+        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(expectedAttributesMap));
+        assertThat(page2Items.get(1).getAttributesMap(), is(expectedAttributesMap));
+        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);
     }
 }