[BEAM-9331] Add better Row builders

sdks/java/core/src/main/java/org/apache/beam/sdk/coders/RowCoderGenerator.java

@@ -359,7 +359,7 @@ static Row decodeDelegate(Schema schema, Coder[] coders, InputStream inputStream
       // all values. Since we assume that decode is always being called on a previously-encoded
       // Row, the values should already be validated and of the correct type. So, we can save
       // some processing by simply transferring ownership of the list to the Row.
-      return Row.withSchema(schema).attachValues(fieldValues).build();
+      return Row.withSchema(schema).attachValues(fieldValues);
     }
   }
 }

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

@@ -610,12 +610,12 @@ private Schema getFieldDescriptorSchema(FieldDescriptor fieldDescriptor, Schema
   private static Schema getFieldSchema(FieldType type) {
     if (TypeName.ROW.equals(type.getTypeName())) {
       return type.getRowSchema();
-    } else if (type.getTypeName().isCollectionType()
-        && TypeName.ROW.equals(type.getCollectionElementType().getTypeName())) {
-      return type.getCollectionElementType().getRowSchema();
-    } else if (TypeName.MAP.equals(type.getTypeName())
-        && TypeName.ROW.equals(type.getMapValueType().getTypeName())) {
-      return type.getMapValueType().getRowSchema();
+    } else if (type.getTypeName().isCollectionType()) {
+      return getFieldSchema(type.getCollectionElementType());
+    } else if (TypeName.MAP.equals(type.getTypeName())) {
+      return getFieldSchema(type.getMapValueType());
+    } else if (TypeName.LOGICAL_TYPE.equals(type.getTypeName())) {
+      return getFieldSchema(type.getLogicalType().getBaseType());
     } else {
       throw new IllegalArgumentException(
           "FieldType " + type + " must be either a row or a container containing rows");

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

@@ -56,7 +56,7 @@ public ToRowWithValueGetters(Schema schema) {
 
     @Override
     public Row apply(T input) {
-      return Row.withSchema(schema).withFieldValueGetters(getterFactory, input).build();
+      return Row.withSchema(schema).withFieldValueGetters(getterFactory, input);
     }
 
     private GetterBasedSchemaProvider getOuter() {

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

@@ -373,7 +373,7 @@ private static Row fillNewFields(Row row, AddFieldsInformation addFieldsInformat
         }
       }
 
-      return Row.withSchema(outputSchema).attachValues(newValues).build();
+      return Row.withSchema(outputSchema).attachValues(newValues);
     }
 
     private static Object fillNewFields(

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

@@ -548,7 +548,7 @@ void outputUnexpandedRow(Schema outputSchema, OutputReceiver<Row> o) {
       List<Object> fields = Lists.newArrayListWithCapacity(getIterables().size() + 1);
       fields.add(getKey());
       fields.addAll(getIterables());
-      o.output(Row.withSchema(outputSchema).attachValues(fields).build());
+      o.output(Row.withSchema(outputSchema).attachValues(fields));
     }
 
     static void verifyExpandedArgs(JoinInformation joinInformation, JoinArguments joinArgs) {
@@ -632,9 +632,7 @@ private void crossProductHelper(
       if (atBottom) {
         // Bottom of recursive call, so output the row we've accumulated.
         Row row =
-            Row.withSchema(getOutputSchema())
-                .attachValues(Lists.newArrayList(accumulatedRows))
-                .build();
+            Row.withSchema(getOutputSchema()).attachValues(Lists.newArrayList(accumulatedRows));
         o.output(row);
       } else {
         crossProduct(tagIndex + 1, accumulatedRows, iterables, o);

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

@@ -859,8 +859,7 @@ public PCollection<Row> expand(PCollection<InputT> input) {
                     public void process(@Element KV<Row, Iterable<Row>> e, OutputReceiver<Row> o) {
                       o.output(
                           Row.withSchema(outputSchema)
-                              .attachValues(Lists.newArrayList(e.getKey(), e.getValue()))
-                              .build());
+                              .attachValues(Lists.newArrayList(e.getKey(), e.getValue())));
                     }
                   }))
           .setRowSchema(outputSchema);
@@ -1140,8 +1139,7 @@ public void process(@Element KV<Row, Row> element, OutputReceiver<Row> o) {
                       o.output(
                           Row.withSchema(outputSchema)
                               .attachValues(
-                                  Lists.newArrayList(element.getKey(), element.getValue()))
-                              .build());
+                                  Lists.newArrayList(element.getKey(), element.getValue())));
                     }
                   }))
           .setRowSchema(outputSchema);

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

@@ -180,7 +180,7 @@ public PCollection<Row> expand(PCollection<T> input) {
                   new DoFn<T, Row>() {
                     @ProcessElement
                     public void processElement(@Element Row row, OutputReceiver<Row> o) {
-                      o.output(Row.withSchema(outputSchema).attachValues(row.getValues()).build());
+                      o.output(Row.withSchema(outputSchema).attachValues(row.getValues()));
                     }
                   }))
           .setRowSchema(outputSchema);

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

@@ -368,11 +368,6 @@ public ByteCodeAppender appender(final Target implementationTarget) {
                             ElementMatchers.named("attachValues")
                                 .and(ElementMatchers.takesArguments(Object[].class)))
                         .getOnly()),
-                MethodInvocation.invoke(
-                    new ForLoadedType(Row.Builder.class)
-                        .getDeclaredMethods()
-                        .filter(ElementMatchers.named("build"))
-                        .getOnly()),
                 MethodReturn.REFERENCE);
         size = size.aggregate(attachToRow.apply(methodVisitor, implementationContext));
         return new Size(size.getMaximalSize(), localVariables.getTotalNumVariables());