Encode DynamoDB empty list and empty map attributes

sdks/java/io/amazon-web-services2/src/main/java/org/apache/beam/sdk/io/aws2/dynamodb/AttributeValueCoder.java

@@ -89,10 +89,13 @@ public void encode(AttributeValue value, OutputStream outStream) throws IOExcept
     } else if (value.bs() != null && value.bs().size() > 0) {
       StringUtf8Coder.of().encode(AttributeValueType.bs.toString(), outStream);
       LIST_BYTE_CODER.encode(convertToListByteArray(value.bs()), outStream);
-    } else if (value.l() != null && value.l().size() > 0) {
+    } else if (value.hasL()) {
+      // hasL/hasM rather than a size check: DynamoDB allows an empty L or M, and the size guard
+      // sent those to the terminal else. The ss/ns/bs guards below keep their size check on
+      // purpose -- DynamoDB rejects empty sets, so an empty one there is not a value to preserve.
       StringUtf8Coder.of().encode(AttributeValueType.l.toString(), outStream);
       LIST_ATTRIBUTE_CODER.encode(value.l(), outStream);
-    } else if (value.m() != null && value.m().size() > 0) {
+    } else if (value.hasM()) {
       StringUtf8Coder.of().encode(AttributeValueType.m.toString(), outStream);
       MAP_ATTRIBUTE_CODER.encode(value.m(), outStream);
     } else if (value.nul() != null) {

sdks/java/io/amazon-web-services2/src/test/java/org/apache/beam/sdk/io/aws2/dynamodb/AttributeValueCoderTest.java

@@ -25,6 +25,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import org.apache.beam.sdk.coders.CoderException;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
 import org.junit.Assert;
 import org.junit.Test;
@@ -205,4 +206,68 @@ public void shouldPassForNullType() throws IOException {
 
     Assert.assertEquals(expected, actual);
   }
+
+  @Test
+  public void shouldPassForEmptyListType() throws IOException {
+    AttributeValue expected = AttributeValue.builder().l(new ArrayList<>()).build();
+
+    AttributeValue actual = roundTrip(expected);
+
+    Assert.assertEquals(expected, actual);
+    Assert.assertTrue("an empty L must decode back as a set-but-empty list", actual.hasL());
+  }
+
+  @Test
+  public void shouldPassForEmptyMapType() throws IOException {
+    AttributeValue expected = AttributeValue.builder().m(new HashMap<>()).build();
+
+    AttributeValue actual = roundTrip(expected);
+
+    Assert.assertEquals(expected, actual);
+    Assert.assertTrue("an empty M must decode back as a set-but-empty map", actual.hasM());
+  }
+
+  @Test
+  public void shouldPassForEmptyListNestedInMap() throws IOException {
+    // The realistic trigger: one empty list anywhere inside an item makes the whole item
+    // unencodable, not just that attribute.
+    Map<String, AttributeValue> item = new HashMap<>();
+    item.put("name", AttributeValue.builder().s("widget").build());
+    item.put("tags", AttributeValue.builder().l(new ArrayList<>()).build());
+    AttributeValue expected = AttributeValue.builder().m(item).build();
+
+    AttributeValue actual = roundTrip(expected);
+
+    Assert.assertEquals(expected, actual);
+    Assert.assertTrue(actual.m().get("tags").hasL());
+  }
+
+  @Test
+  public void shouldPassForEmptyMapNestedInList() throws IOException {
+    AttributeValue expected =
+        AttributeValue.builder()
+            .l(ImmutableList.of(AttributeValue.builder().m(new HashMap<>()).build()))
+            .build();
+
+    AttributeValue actual = roundTrip(expected);
+
+    Assert.assertEquals(expected, actual);
+    Assert.assertTrue(actual.l().get(0).hasM());
+  }
+
+  @Test
+  public void shouldStillRejectAnAttributeValueWithNoTypeSet() throws IOException {
+    // Control: the terminal else must keep rejecting a genuinely typeless value, so the fix does
+    // not turn "Unknown Type" into silently encoding nothing.
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    Assert.assertThrows(
+        CoderException.class,
+        () -> AttributeValueCoder.of().encode(AttributeValue.builder().build(), out));
+  }
+
+  private static AttributeValue roundTrip(AttributeValue value) throws IOException {
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    AttributeValueCoder.of().encode(value, out);
+    return AttributeValueCoder.of().decode(new ByteArrayInputStream(out.toByteArray()));
+  }
 }