Fix inconsistent data type in GenericRecord and AvroSchema for AvroWriter

sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java

@@ -1214,6 +1214,16 @@ private static org.apache.avro.Schema getFieldSchema(
     return fieldType.getNullable() ? ReflectData.makeNullable(baseType) : baseType;
   }
 
+  private static final org.apache.avro.Schema INT_AVRO_TYPE =
+      org.apache.avro.Schema.create(Type.INT);
+  private static final org.apache.avro.Schema LONG_AVRO_TYPE =
+      org.apache.avro.Schema.create(Type.LONG);
+  private static final org.apache.avro.Schema FLOAT_AVRO_TYPE =
+      org.apache.avro.Schema.create(Type.FLOAT);
+  private static final org.apache.avro.Schema DOUBLE_AVRO_TYPE =
+      org.apache.avro.Schema.create(Type.DOUBLE);
+
+  /** Convert a value from Beam Row to a vlue used for Avro GenericRecord. */
   private static @Nullable Object genericFromBeamField(
       FieldType fieldType, org.apache.avro.Schema avroSchema, @Nullable Object value) {
     TypeWithNullability typeWithNullability = new TypeWithNullability(avroSchema);
@@ -1230,6 +1240,18 @@ private static org.apache.avro.Schema getFieldSchema(
       return value;
     }
 
+    // Handle implicit up-cast: use avroSchema
+    if (INT_AVRO_TYPE.equals(typeWithNullability.type)) {
+      return ((Number) value).intValue();
+    } else if (LONG_AVRO_TYPE.equals(typeWithNullability.type)) {
+      return ((Number) value).longValue();
+    } else if (FLOAT_AVRO_TYPE.equals(typeWithNullability.type)) {
+      return ((Number) value).floatValue();
+    } else if (DOUBLE_AVRO_TYPE.equals(typeWithNullability.type)) {
+      return ((Number) value).doubleValue();
+    }
+
+    // TODO: should we use Avro Schema as the source-of-truth in general?
     switch (fieldType.getTypeName()) {
       case BYTE:
       case INT16:

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

@@ -122,7 +122,7 @@ public SerializableFunction<AvroWriteRequest<Row>, GenericRecord> getAvroFilterF
         row = checkStateNotNull(row.getRow(RECORD));
       }
       Row filtered = rowFilter.filter(row);
-      return AvroUtils.toGenericRecord(filtered);
+      return AvroUtils.toGenericRecord(filtered, request.getSchema());
     };
   }
 }

sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/providers/BigQueryFileLoadsSchemaTransformProviderTest.java

@@ -70,13 +70,28 @@ public class BigQueryFileLoadsSchemaTransformProviderTest {
       new TableReference().setProjectId(PROJECT).setDatasetId(DATASET).setTableId(TABLE_ID);
 
   private static final Schema SCHEMA =
-      Schema.of(Field.of("name", FieldType.STRING), Field.of("number", FieldType.INT64));
+      Schema.of(
+          Field.of("name", FieldType.STRING),
+          Field.of("number", FieldType.INT64),
+          Field.of("age", FieldType.INT32).withNullable(true));
 
   private static final List<Row> ROWS =
       Arrays.asList(
-          Row.withSchema(SCHEMA).withFieldValue("name", "a").withFieldValue("number", 1L).build(),
-          Row.withSchema(SCHEMA).withFieldValue("name", "b").withFieldValue("number", 2L).build(),
-          Row.withSchema(SCHEMA).withFieldValue("name", "c").withFieldValue("number", 3L).build());
+          Row.withSchema(SCHEMA)
+              .withFieldValue("name", "a")
+              .withFieldValue("number", 1L)
+              .withFieldValue("age", 10)
+              .build(),
+          Row.withSchema(SCHEMA)
+              .withFieldValue("name", "b")
+              .withFieldValue("number", 2L)
+              .withFieldValue("age", 20)
+              .build(),
+          Row.withSchema(SCHEMA)
+              .withFieldValue("name", "c")
+              .withFieldValue("number", 3L)
+              .withFieldValue("age", null)
+              .build());
 
   private static final BigQueryOptions OPTIONS =
       TestPipeline.testingPipelineOptions().as(BigQueryOptions.class);