tableRowFromMessage should use setF

Author: reuvenlax

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowToStorageApiProto.java

@@ -1129,10 +1129,12 @@ private static long toEpochMicros(Instant timestamp) {
   @VisibleForTesting
   public static TableRow tableRowFromMessage(
       Message message, boolean includeCdcColumns, Predicate<String> includeField) {
-    return tableRowFromMessage(message, includeCdcColumns, includeField, "");
+    Optional<TableRow> tableRow = tableRowFromMessage(message, includeCdcColumns, includeField, "");
+    return tableRow.orElseGet(
+        () -> tableRowFromMessageUseSetF(message, includeCdcColumns, includeField, ""));
   }
 
-  public static TableRow tableRowFromMessage(
+  public static Optional<TableRow> tableRowFromMessage(
       Message message,
       boolean includeCdcColumns,
       Predicate<String> includeField,
@@ -1143,36 +1145,113 @@ public static TableRow tableRowFromMessage(
       StringBuilder fullName = new StringBuilder();
       FieldDescriptor fieldDescriptor = field.getKey();
       String fieldName = fieldNameFromProtoFieldDescriptor(fieldDescriptor);
+      if ("f".equals(fieldName)) {
+        return Optional.empty();
+      }
       fullName = fullName.append(namePrefix).append(fieldName);
       Object fieldValue = field.getValue();
       if ((includeCdcColumns || !StorageApiCDC.COLUMNS.contains(fullName.toString()))
           && includeField.test(fieldName)) {
-        tableRow.put(
-            fieldName,
+        Object convertedFieldValue =
             jsonValueFromMessageValue(
-                fieldDescriptor, fieldValue, true, includeField, fullName.append(".").toString()));
+                fieldDescriptor,
+                fieldValue,
+                true,
+                includeField,
+                fullName.append(".").toString(),
+                false);
+        if (convertedFieldValue instanceof Optional) {
+          Optional<?> optional = (Optional<?>) convertedFieldValue;
+          if (!optional.isPresent()) {
+            return Optional.empty();
+          } else {
+            convertedFieldValue = optional.get();
+          }
+        }
+        tableRow.put(fieldName, convertedFieldValue);
       }
     }
+    return Optional.of(tableRow);
+  }
+
+  public static TableRow tableRowFromMessageUseSetF(
+      Message message,
+      boolean includeCdcColumns,
+      Predicate<String> includeField,
+      String namePrefix) {
+    List<TableCell> tableCells =
+        Lists.newArrayListWithCapacity(message.getDescriptorForType().getFields().size());
+
+    for (FieldDescriptor fieldDescriptor : message.getDescriptorForType().getFields()) {
+      TableCell tableCell = new TableCell();
+
+      boolean isPresent =
+          (fieldDescriptor.isRepeated() && message.getRepeatedFieldCount(fieldDescriptor) > 0)
+              || (!fieldDescriptor.isRepeated() && message.hasField(fieldDescriptor));
+      if (isPresent) {
+        StringBuilder fullName = new StringBuilder();
+        String fieldName = fieldNameFromProtoFieldDescriptor(fieldDescriptor);
+        fullName = fullName.append(namePrefix).append(fieldName);
+        if ((includeCdcColumns || !StorageApiCDC.COLUMNS.contains(fullName.toString()))
+            && includeField.test(fieldName)) {
+          Object fieldValue = message.getField(fieldDescriptor);
+          Object converted =
+              jsonValueFromMessageValue(
+                  fieldDescriptor,
+                  fieldValue,
+                  true,
+                  includeField,
+                  fullName.append(".").toString(),
+                  true);
+          tableCell.setV(converted);
+        }
+      }
+      tableCells.add(tableCell);
+    }
+
+    TableRow tableRow = new TableRow();
+    tableRow.setF(tableCells);
+
     return tableRow;
   }
 
+  // Translate a proto message value into a json value. If useSetF==false, this will fail with
+  // Optional.empty() if
+  // any fields named "f" are found (due to restrictions on the TableRow class). In that case, the
+  // top level will retry
+  // with useSetF==true. We fallback this way in order to maintain backwards compatibility with
+  // existing users.
   public static Object jsonValueFromMessageValue(
       FieldDescriptor fieldDescriptor,
       Object fieldValue,
       boolean expandRepeated,
       Predicate<String> includeField,
-      String prefix) {
+      String prefix,
+      boolean useSetF) {
     if (expandRepeated && fieldDescriptor.isRepeated()) {
       List<Object> valueList = (List<Object>) fieldValue;
-      return valueList.stream()
-          .map(v -> jsonValueFromMessageValue(fieldDescriptor, v, false, includeField, prefix))
-          .collect(toList());
+      List<Object> expanded = Lists.newArrayListWithCapacity(valueList.size());
+      for (Object value : valueList) {
+        Object translatedValue =
+            jsonValueFromMessageValue(fieldDescriptor, value, false, includeField, prefix, useSetF);
+        if (!useSetF && translatedValue instanceof Optional) {
+          Optional<?> optional = (Optional<?>) translatedValue;
+          if (!optional.isPresent()) {
+            return Optional.empty();
+          }
+          translatedValue = optional.get();
+        }
+        expanded.add(translatedValue);
+      }
+      return expanded;
     }
 
     switch (fieldDescriptor.getType()) {
       case GROUP:
       case MESSAGE:
-        return tableRowFromMessage((Message) fieldValue, false, includeField, prefix);
+        return useSetF
+            ? tableRowFromMessageUseSetF((Message) fieldValue, false, includeField, prefix)
+            : tableRowFromMessage((Message) fieldValue, false, includeField, prefix);
       case BYTES:
         return BaseEncoding.base64().encode(((ByteString) fieldValue).toByteArray());
       case ENUM: