Fix Row.toString NPE on a null nested inside an array, map or row

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaUtils.java

@@ -238,6 +238,13 @@ static String toPrettyOptionsString(Schema.Options options, String prefix) {
   }
 
   static String toPrettyFieldValueString(Schema.FieldType fieldType, Object value, String prefix) {
+    if (value == null) {
+      // toPrettyRowString drops a row's own null fields, but a null nested inside an array, an
+      // iterable, a map or a row has a position and has to be rendered. The numeric and boolean
+      // branches below already render one as "null" via Objects.toString, so do the same for the
+      // remaining types rather than dereferencing the value.
+      return "null";
+    }
     String nextPrefix = prefix + INDENT;
     switch (fieldType.getTypeName()) {
       case BYTE:

sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/SchemaUtilsTest.java

@@ -18,8 +18,13 @@
 package org.apache.beam.sdk.schemas;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
 import org.apache.beam.sdk.schemas.Schema.FieldType;
+import org.apache.beam.sdk.values.Row;
 import org.junit.Test;
 
 /** Tests for {@link org.apache.beam.sdk.schemas.SchemaUtils}. */
@@ -104,4 +109,42 @@ public void testWidenMap() {
             .build();
     assertEquals(expected, SchemaUtils.mergeWideningNullable(schema1, schema2));
   }
+
+  @Test
+  public void testToPrettyStringRendersNullInsideArray() {
+    Schema schema =
+        Schema.builder().addArrayField("a", FieldType.STRING.withNullable(true)).build();
+    Row row = Row.withSchema(schema).addValue(Arrays.asList("x", null)).build();
+    assertTrue(row.toString(), row.toString().contains("null"));
+  }
+
+  @Test
+  public void testToPrettyStringRendersNullMapValue() {
+    Schema schema =
+        Schema.builder()
+            .addMapField("m", FieldType.STRING, FieldType.STRING.withNullable(true))
+            .build();
+    Map<String, String> map = new HashMap<>();
+    map.put("k", null);
+    Row row = Row.withSchema(schema).addValue(map).build();
+    assertTrue(row.toString(), row.toString().contains("null"));
+  }
+
+  @Test
+  public void testToPrettyStringRendersNullRowInsideArray() {
+    Schema inner = Schema.builder().addStringField("s").build();
+    Schema schema =
+        Schema.builder().addArrayField("a", FieldType.row(inner).withNullable(true)).build();
+    Row row = Row.withSchema(schema).addValue(Arrays.asList((Row) null)).build();
+    assertTrue(row.toString(), row.toString().contains("null"));
+  }
+
+  @Test
+  public void testToPrettyStringRendersNullInsideArrayOfInts() {
+    Schema schema = Schema.builder().addArrayField("a", FieldType.INT32.withNullable(true)).build();
+    Row row = Row.withSchema(schema).addValue(Arrays.asList(1, null)).build();
+    String rendered = row.toString();
+    assertTrue(rendered, rendered.contains("1"));
+    assertTrue(rendered, rendered.contains("null"));
+  }
 }