fix: update schema compat check with backend type changes (#522)

* fix: update schema compat check with backend type changes

* remove BIGNUMERIC, the client library is not populated with the type

Author: yirutang

google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1alpha2/SchemaCompatibility.java

@@ -156,8 +156,8 @@ private static boolean isCompatibleWithBQDate(Descriptors.FieldDescriptor.Type f
   }
 
   private static boolean isCompatibleWithBQDatetime(Descriptors.FieldDescriptor.Type field) {
-    if (field == Descriptors.FieldDescriptor.Type.INT64
-        || field == Descriptors.FieldDescriptor.Type.SFIXED64) {
+    if (field == Descriptors.FieldDescriptor.Type.STRING
+        || field == Descriptors.FieldDescriptor.Type.INT64) {
       return true;
     }
     return false;
@@ -174,7 +174,7 @@ private static boolean isCompatibleWithBQFloat(Descriptors.FieldDescriptor.Type
   }
 
   private static boolean isCompatibleWithBQGeography(Descriptors.FieldDescriptor.Type field) {
-    if (field == Descriptors.FieldDescriptor.Type.BYTES) {
+    if (field == Descriptors.FieldDescriptor.Type.STRING) {
       return true;
     }
     return false;
@@ -201,15 +201,10 @@ private static boolean isCompatibleWithBQNumeric(Descriptors.FieldDescriptor.Typ
         || field == Descriptors.FieldDescriptor.Type.FIXED32
         || field == Descriptors.FieldDescriptor.Type.FIXED64
         || field == Descriptors.FieldDescriptor.Type.SFIXED32
-        || field == Descriptors.FieldDescriptor.Type.SFIXED64) {
-      return true;
-    }
-
-    if (field == Descriptors.FieldDescriptor.Type.BYTES) {
-      return true;
-    }
-
-    if (field == Descriptors.FieldDescriptor.Type.FLOAT
+        || field == Descriptors.FieldDescriptor.Type.SFIXED64
+        || field == Descriptors.FieldDescriptor.Type.STRING
+        || field == Descriptors.FieldDescriptor.Type.BYTES
+        || field == Descriptors.FieldDescriptor.Type.FLOAT
         || field == Descriptors.FieldDescriptor.Type.DOUBLE) {
       return true;
     }
@@ -235,7 +230,7 @@ private static boolean isCompatibleWithBQString(Descriptors.FieldDescriptor.Type
 
   private static boolean isCompatibleWithBQTime(Descriptors.FieldDescriptor.Type field) {
     if (field == Descriptors.FieldDescriptor.Type.INT64
-        || field == Descriptors.FieldDescriptor.Type.SFIXED64) {
+        || field == Descriptors.FieldDescriptor.Type.STRING) {
 
       return true;
     }

google-cloud-bigquerystorage/src/test/java/com/google/cloud/bigquery/storage/v1alpha2/SchemaCompatibilityTest.java

@@ -587,7 +587,7 @@ public void testBQDatetime() {
                 .build()));
     SchemaCompatibility compact = SchemaCompatibility.getInstance(mockBigquery);
     HashSet<Descriptors.Descriptor> compatible =
-        new HashSet<>(Arrays.asList(Int64Type.getDescriptor(), SFixed64Type.getDescriptor()));
+        new HashSet<>(Arrays.asList(Int64Type.getDescriptor(), StringType.getDescriptor()));
 
     for (Descriptors.Descriptor descriptor : type_descriptors) {
       if (compatible.contains(descriptor)) {
@@ -649,7 +649,7 @@ public void testBQGeography() {
                 .build()));
     SchemaCompatibility compact = SchemaCompatibility.getInstance(mockBigquery);
     HashSet<Descriptors.Descriptor> compatible =
-        new HashSet<>(Arrays.asList(BytesType.getDescriptor()));
+        new HashSet<>(Arrays.asList(StringType.getDescriptor()));
 
     for (Descriptors.Descriptor descriptor : type_descriptors) {
       if (compatible.contains(descriptor)) {
@@ -730,6 +730,7 @@ public void testBQNumeric() {
                 SFixed32Type.getDescriptor(),
                 SFixed64Type.getDescriptor(),
                 BytesType.getDescriptor(),
+                StringType.getDescriptor(),
                 FloatType.getDescriptor(),
                 DoubleType.getDescriptor()));
 
@@ -879,7 +880,7 @@ public void testBQTime() {
                 .build()));
     SchemaCompatibility compact = SchemaCompatibility.getInstance(mockBigquery);
     HashSet<Descriptors.Descriptor> compatible =
-        new HashSet<>(Arrays.asList(Int64Type.getDescriptor(), SFixed64Type.getDescriptor()));
+        new HashSet<>(Arrays.asList(Int64Type.getDescriptor(), StringType.getDescriptor()));
 
     for (Descriptors.Descriptor descriptor : type_descriptors) {
       if (compatible.contains(descriptor)) {