[beam] branch master updated: Avro datum factory fixup (#27173)

[yh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

yhu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new 4c66866aa95 Avro datum factory fixup (#27173)
4c66866aa95 is described below

commit 4c66866aa9544d1796c7c3880192cb57d2a8dcc0
Author: Michel Davit <mi...@davit.fr>
AuthorDate: Tue Jun 20 17:08:17 2023 +0200

    Avro datum factory fixup (#27173)
    
    * Remove DecimalConversion from logical-type conversions
    
    * Restore AvroCloudObject compatibility
---
 .../util/AvroCoderCloudObjectTranslator.java       | 30 +++++++++++++++++-----
 .../extensions/avro/schemas/utils/AvroUtils.java   |  3 ++-
 2 files changed, 25 insertions(+), 8 deletions(-)

diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/AvroCoderCloudObjectTranslator.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/AvroCoderCloudObjectTranslator.java
index e5c6ded2fdf..9f5913f051b 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/AvroCoderCloudObjectTranslator.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/AvroCoderCloudObjectTranslator.java
@@ -31,6 +31,9 @@ import org.apache.beam.sdk.util.StringUtils;
 class AvroCoderCloudObjectTranslator implements CloudObjectTranslator<AvroCoder> {
   private static final String DATUM_FACTORY_FIELD = "datum_factory";
   private static final String SCHEMA_FIELD = "schema";
+  // deprecated fields
+  private static final String TYPE_FIELD = "type";
+  private static final String REFLECT_API_FIELD = "reflect_api";
 
   @Override
   public CloudObject toCloudObject(AvroCoder target, SdkComponents sdkComponents) {
@@ -45,15 +48,28 @@ class AvroCoderCloudObjectTranslator implements CloudObjectTranslator<AvroCoder>
 
   @Override
   public AvroCoder<?> fromCloudObject(CloudObject cloudObject) {
-    Schema.Parser parser = new Schema.Parser();
-    byte[] deserializedDatumFactory =
-        StringUtils.jsonStringToByteArray(Structs.getString(cloudObject, DATUM_FACTORY_FIELD));
     String schemaString = Structs.getString(cloudObject, SCHEMA_FIELD);
+    Schema.Parser parser = new Schema.Parser();
     Schema schema = parser.parse(schemaString);
-    AvroDatumFactory<?> datumFactory =
-        (AvroDatumFactory)
-            SerializableUtils.deserializeFromByteArray(
-                deserializedDatumFactory, DATUM_FACTORY_FIELD);
+    AvroDatumFactory<?> datumFactory;
+    if (cloudObject.containsKey(TYPE_FIELD)) {
+      // coder was created with an older beam version. use default datum factory
+      try {
+        String className = Structs.getString(cloudObject, TYPE_FIELD);
+        Class<?> type = Class.forName(className);
+        boolean useReflectApi = Structs.getBoolean(cloudObject, REFLECT_API_FIELD);
+        datumFactory = AvroDatumFactory.of(type, useReflectApi);
+      } catch (ClassNotFoundException e) {
+        throw new IllegalArgumentException(e);
+      }
+    } else {
+      byte[] deserializedDatumFactory =
+          StringUtils.jsonStringToByteArray(Structs.getString(cloudObject, DATUM_FACTORY_FIELD));
+      datumFactory =
+          (AvroDatumFactory)
+              SerializableUtils.deserializeFromByteArray(
+                  deserializedDatumFactory, DATUM_FACTORY_FIELD);
+    }
     return AvroCoder.of(datumFactory, schema);
   }
 
diff --git a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java
index eb1b9cdaba4..1f4fca5e831 100644
--- a/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java
+++ b/sdks/java/extensions/avro/src/main/java/org/apache/beam/sdk/extensions/avro/schemas/utils/AvroUtils.java
@@ -153,7 +153,8 @@ public class AvroUtils {
   private static final ForLoadedType JODA_INSTANT = new ForLoadedType(Instant.class);
 
   public static void addLogicalTypeConversions(final GenericData data) {
-    data.addLogicalTypeConversion(new Conversions.DecimalConversion());
+    // do not add DecimalConversion by default as schema must have extra 'scale' and 'precision'
+    // properties. avro reflect already handles BigDecimal as string with the 'java-class' property
     data.addLogicalTypeConversion(new Conversions.UUIDConversion());
     // joda-time
     data.addLogicalTypeConversion(new AvroJodaTimeConversions.DateConversion());