better merge (feast-dev#945)

Author: Oleksii Moskalenko

storage/connectors/bigquery/src/main/java/feast/storage/connectors/bigquery/writer/FeatureSetSpecToTableSchema.java

@@ -28,9 +28,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.stream.Collectors;
 import org.apache.beam.repackaged.core.org.apache.commons.lang3.tuple.Pair;
 import org.apache.beam.sdk.coders.Coder;
@@ -146,8 +144,11 @@ private void createTable(String specKey, Schema schema) {
    */
   private Schema createSchemaFromSpec(
       FeatureSetProto.FeatureSetSpec spec, String specKey, Table existingTable) {
-    List<Field> fields = new ArrayList<>();
-    log.info("Table {} will have the following fields:", specKey);
+    Map<String, Field> fields = new LinkedHashMap<>();
+    if (existingTable != null) {
+      Schema existingSchema = existingTable.getDefinition().getSchema();
+      existingSchema.getFields().forEach(f -> fields.put(f.getName(), f));
+    }
 
     for (FeatureSetProto.EntitySpec entitySpec : spec.getEntitiesList()) {
       Field.Builder builder =
@@ -157,8 +158,7 @@ private Schema createSchemaFromSpec(
         builder.setMode(Field.Mode.REPEATED);
       }
       Field field = builder.build();
-      log.info("- {}", field.toString());
-      fields.add(field);
+      fields.put(field.getName(), field);
     }
     for (FeatureSetProto.FeatureSpec featureSpec : spec.getFeaturesList()) {
       Field.Builder builder =
@@ -169,8 +169,7 @@ private Schema createSchemaFromSpec(
       }
 
       Field field = builder.build();
-      log.info("- {}", field.toString());
-      fields.add(field);
+      fields.put(field.getName(), field);
     }
 
     // Refer to protos/feast/core/Store.proto for reserved fields in BigQuery.
@@ -192,24 +191,13 @@ private Schema createSchemaFromSpec(
           Field.newBuilder(entry.getKey(), entry.getValue().getLeft())
               .setDescription(entry.getValue().getRight())
               .build();
-      log.info("- {}", field.toString());
-      fields.add(field);
+      fields.put(field.getName(), field);
     }
 
-    List<Field> fieldsList = new ArrayList<>();
-    if (existingTable != null) {
-      Schema existingSchema = existingTable.getDefinition().getSchema();
-      fieldsList.addAll(existingSchema.getFields());
-    }
-
-    for (Field field : fields) {
-      if (!fieldsList.contains(field)) {
-        fieldsList.add(field);
-        log.info("- {}", field.toString());
-      }
-    }
+    log.info("Table {} will have the following fields:", specKey);
+    fields.values().forEach(f -> log.info("- {}", f.toString()));
 
-    return Schema.of(FieldList.of(fieldsList));
+    return Schema.of(FieldList.of(fields.values()));
   }
 
   /**

storage/connectors/bigquery/src/test/java/feast/storage/connectors/bigquery/writer/BigQuerySinkTest.java

@@ -315,7 +315,9 @@ public void updateSchemaWithExistingTable() {
                 TableId.of("test-project", "test_dataset", "myproject_fs_2"),
                 StandardTableDefinition.of(
                     Schema.of(
-                        Field.of("old_feature_1", LegacySQLTypeName.FLOAT),
+                        Field.newBuilder("old_feature_1", LegacySQLTypeName.FLOAT)
+                            .setDescription("Some old description")
+                            .build(),
                         Field.of("old_feature_2", LegacySQLTypeName.INTEGER)))));
 
     FeatureSetSpec spec_fs_2 =
@@ -332,6 +334,11 @@ public void updateSchemaWithExistingTable() {
                     .setName("feature")
                     .setValueType(ValueProto.ValueType.Enum.STRING)
                     .build())
+            .addFeatures(
+                FeatureSpec.newBuilder()
+                    .setName("old_feature_1")
+                    .setValueType(ValueProto.ValueType.Enum.FLOAT)
+                    .build())
             .build();
 
     FeatureSink sink =