tableRowFromMessage should use setF

Author: reuvenlax

sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumReadSchemaTransformProvider.java

@@ -23,7 +23,6 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
-import java.util.Objects;
 import java.util.stream.Collectors;
 import org.apache.beam.sdk.coders.RowCoder;
 import org.apache.beam.sdk.schemas.Schema;

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

@@ -1137,22 +1137,33 @@ public static TableRow tableRowFromMessage(
       boolean includeCdcColumns,
       Predicate<String> includeField,
       String namePrefix) {
-    // TODO: Would be more correct to generate TableRows using setF.
-    TableRow tableRow = new TableRow();
-    for (Map.Entry<FieldDescriptor, Object> field : message.getAllFields().entrySet()) {
-      StringBuilder fullName = new StringBuilder();
-      FieldDescriptor fieldDescriptor = field.getKey();
-      String fieldName = fieldNameFromProtoFieldDescriptor(fieldDescriptor);
-      fullName = fullName.append(namePrefix).append(fieldName);
-      Object fieldValue = field.getValue();
-      if ((includeCdcColumns || !StorageApiCDC.COLUMNS.contains(fullName.toString()))
-          && includeField.test(fieldName)) {
-        tableRow.put(
-            fieldName,
-            jsonValueFromMessageValue(
-                fieldDescriptor, fieldValue, true, includeField, fullName.append(".").toString()));
+    List<TableCell> tableCells =
+        Lists.newArrayListWithCapacity(message.getDescriptorForType().getFields().size());
+
+    for (FieldDescriptor fieldDescriptor : message.getDescriptorForType().getFields()) {
+      TableCell tableCell = new TableCell();
+      if (message.hasField(fieldDescriptor)) {
+        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);
+          tableCell.setV(
+              jsonValueFromMessageValue(
+                  fieldDescriptor,
+                  fieldValue,
+                  true,
+                  includeField,
+                  fullName.append(".").toString()));
+        }
       }
+      tableCells.add(tableCell);
     }
+
+    TableRow tableRow = new TableRow();
+    tableRow.setF(tableCells);
+
     return tableRow;
   }