Render ARRAY and ITERABLE fields that are not backed by a List

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

@@ -18,7 +18,6 @@
 package org.apache.beam.sdk.schemas;
 
 import java.util.Arrays;
-import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import org.apache.beam.sdk.schemas.Schema.FieldType;
@@ -270,7 +269,11 @@ static String toPrettyFieldValueString(Schema.FieldType fieldType, Object value,
       case ARRAY:
       case ITERABLE:
         {
-          if (!(value instanceof List)) {
+          // An ITERABLE field declares an Iterable, and an ARRAY value need not be a List
+          // either, so require only what is actually iterated below. This is reached from
+          // Row#toString, where refusing to render one field takes out logging and debugger
+          // output for the entire row.
+          if (!(value instanceof Iterable)) {
             throw new IllegalArgumentException(
                 String.format(
                     "value type is '%s' for field type '%s'",
@@ -279,24 +282,23 @@ static String toPrettyFieldValueString(Schema.FieldType fieldType, Object value,
           FieldType elementType = Objects.requireNonNull(fieldType.getCollectionElementType());
 
           @SuppressWarnings("unchecked")
-          List<Object> list = (List<Object>) value;
-          if (list.isEmpty()) {
-            return "[]";
-          }
+          Iterable<Object> elements = (Iterable<Object>) value;
+
           StringBuilder sb = new StringBuilder();
           sb.append("[\n");
-          int size = list.size();
-          int index = 0;
-          for (Object element : list) {
-            sb.append(nextPrefix)
-                .append(toPrettyFieldValueString(elementType, element, nextPrefix));
-            if (index++ < size - 1) {
+          boolean empty = true;
+          for (Object element : elements) {
+            if (!empty) {
               sb.append(",\n");
-            } else {
-              sb.append("\n");
             }
+            sb.append(nextPrefix)
+                .append(toPrettyFieldValueString(elementType, element, nextPrefix));
+            empty = false;
+          }
+          if (empty) {
+            return "[]";
           }
-          sb.append(prefix).append("]");
+          sb.append("\n").append(prefix).append("]");
           return sb.toString();
         }
       case MAP:

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

@@ -21,7 +21,9 @@
 import static org.junit.Assert.assertTrue;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.Map;
 import org.apache.beam.sdk.schemas.Schema.FieldType;
 import org.apache.beam.sdk.values.Row;
@@ -147,4 +149,69 @@ public void testToPrettyStringRendersNullInsideArrayOfInts() {
     assertTrue(rendered, rendered.contains("1"));
     assertTrue(rendered, rendered.contains("null"));
   }
+
+  @Test
+  public void testToPrettyStringRendersIterableThatIsNotAList() {
+    Schema schema =
+        Schema.builder().addStringField("k").addIterableField("vals", FieldType.STRING).build();
+    // A bare Iterable, which is exactly what an ITERABLE field declares.
+    Iterable<String> bare = () -> Arrays.asList("p", "q").iterator();
+    Row row = Row.withSchema(schema).attachValues("k1", bare);
+
+    // The precondition that makes this test meaningful: if the value ever starts arriving as a
+    // List, the assertion below would pass for the wrong reason.
+    assertTrue(!(row.getValue("vals") instanceof java.util.List));
+
+    String rendered = row.toString();
+    assertTrue(rendered, rendered.contains("p"));
+    assertTrue(rendered, rendered.contains("q"));
+  }
+
+  @Test
+  public void testToPrettyStringStillRendersAListArray() {
+    // Control: the ordinary List case must be unchanged.
+    Schema schema = Schema.builder().addArrayField("vals", FieldType.STRING).build();
+    Row row = Row.withSchema(schema).attachValues((Object) Arrays.asList("p", "q"));
+
+    String rendered = row.toString();
+    assertTrue(rendered, rendered.contains("p"));
+    assertTrue(rendered, rendered.contains("q"));
+  }
+
+  @Test
+  public void testToPrettyStringRendersAnIterableThatCanOnlyBeReadOnce() {
+    Schema schema =
+        Schema.builder().addStringField("k").addIterableField("vals", FieldType.STRING).build();
+    // Hands out its iterator exactly once and fails loudly on a second attempt. Rendering asks
+    // the collection for one traversal now; asking for isEmpty(), then size(), then iterating
+    // would trip the guard below.
+    Iterable<String> onceOnly =
+        new Iterable<String>() {
+          private boolean taken = false;
+
+          @Override
+          public Iterator<String> iterator() {
+            if (taken) {
+              throw new IllegalStateException("iterated more than once");
+            }
+            taken = true;
+            return Arrays.asList("p", "q").iterator();
+          }
+        };
+    Row row = Row.withSchema(schema).attachValues("k1", onceOnly);
+
+    String rendered = row.toString();
+    assertTrue(rendered, rendered.contains("p"));
+    assertTrue(rendered, rendered.contains("q"));
+  }
+
+  @Test
+  public void testToPrettyStringRendersAnEmptyIterableAsEmptyBrackets() {
+    Schema schema = Schema.builder().addIterableField("vals", FieldType.STRING).build();
+    Iterable<String> empty = Collections::emptyIterator;
+    Row row = Row.withSchema(schema).attachValues((Object) empty);
+
+    String rendered = row.toString();
+    assertTrue(rendered, rendered.contains("[]"));
+  }
 }