Add feature and feature set labels, for metadata

core/src/main/java/feast/core/model/FeatureSet.java

@@ -25,6 +25,7 @@
 import feast.core.FeatureSetProto.FeatureSetSpec;
 import feast.core.FeatureSetProto.FeatureSetStatus;
 import feast.core.FeatureSetProto.FeatureSpec;
+import feast.core.util.TypeConversion;
 import feast.types.ValueProto.ValueType.Enum;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -116,6 +117,10 @@ public class FeatureSet extends AbstractTimestampEntity implements Comparable<Fe
   @Column(name = "status")
   private String status;
 
+  // User defined metadata
+  @Column(name = "labels", columnDefinition = "text")
+  private String labels;
+
   public FeatureSet() {
     super();
   }
@@ -128,6 +133,7 @@ public FeatureSet(
       List<Field> entities,
       List<Field> features,
       Source source,
+      Map<String, String> labels,
       FeatureSetStatus status) {
     this.maxAgeSeconds = maxAgeSeconds;
     this.source = source;
@@ -137,6 +143,7 @@ public FeatureSet(
     this.name = name;
     this.project = new Project(project);
     this.version = version;
+    this.labels = TypeConversion.convertMapToJsonString(labels);
     this.setId(project, name, version);
     addEntities(entities);
     addFeatures(features);
@@ -191,6 +198,7 @@ public static FeatureSet fromProto(FeatureSetProto.FeatureSet featureSetProto) {
         entitySpecs,
         featureSpecs,
         source,
+        featureSetProto.getSpec().getLabelsMap(),
         featureSetProto.getMeta().getStatus());
   }
 
@@ -247,6 +255,7 @@ public FeatureSetProto.FeatureSet toProto() throws InvalidProtocolBufferExceptio
             .setMaxAge(Duration.newBuilder().setSeconds(maxAgeSeconds))
             .addAllEntities(entitySpecs)
             .addAllFeatures(featureSpecs)
+            .putAllLabels(TypeConversion.convertJsonStringToMap(labels))
             .setSource(source.toProto());
 
     return FeatureSetProto.FeatureSet.newBuilder().setMeta(meta).setSpec(spec).build();
@@ -352,6 +361,10 @@ private void setFeatureSpecFields(FeatureSpec.Builder featureSpecBuilder, Field
       featureSpecBuilder.setTimeOfDayDomain(
           TimeOfDayDomain.parseFrom(featureField.getTimeOfDayDomain()));
     }
+
+    if (featureField.getLabels() != null) {
+      featureSpecBuilder.putAllLabels(featureField.getLabels());
+    }
   }
 
   /**

core/src/main/java/feast/core/model/Field.java

@@ -18,8 +18,9 @@
 
 import feast.core.FeatureSetProto.EntitySpec;
 import feast.core.FeatureSetProto.FeatureSpec;
-import feast.types.ValueProto.ValueType;
+import feast.core.util.TypeConversion;
 import java.util.Arrays;
+import java.util.Map;
 import java.util.Objects;
 import javax.persistence.Column;
 import javax.persistence.Embeddable;
@@ -47,6 +48,10 @@ public class Field {
   @Column(name = "project")
   private String project;
 
+  // Labels that this field belongs to
+  @Column(name = "labels", columnDefinition = "text")
+  private String labels;
+
   // Presence constraints (refer to proto feast.core.FeatureSet.FeatureSpec)
   // Only one of them can be set.
   private byte[] presence;
@@ -74,14 +79,10 @@ public class Field {
 
   public Field() {}
 
-  public Field(String name, ValueType.Enum type) {
-    this.name = name;
-    this.type = type.toString();
-  }
-
   public Field(FeatureSpec featureSpec) {
     this.name = featureSpec.getName();
     this.type = featureSpec.getValueType().toString();
+    this.labels = TypeConversion.convertMapToJsonString(featureSpec.getLabelsMap());
 
     switch (featureSpec.getPresenceConstraintsCase()) {
       case PRESENCE:
@@ -215,6 +216,10 @@ public Field(EntitySpec entitySpec) {
     }
   }
 
+  public Map<String, String> getLabels() {
+    return TypeConversion.convertJsonStringToMap(this.labels);
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {
@@ -227,6 +232,7 @@ public boolean equals(Object o) {
     return Objects.equals(name, field.name)
         && Objects.equals(type, field.type)
         && Objects.equals(project, field.project)
+        && Objects.equals(labels, field.labels)
         && Arrays.equals(presence, field.presence)
         && Arrays.equals(groupPresence, field.groupPresence)
         && Arrays.equals(shape, field.shape)
@@ -247,6 +253,6 @@ public boolean equals(Object o) {
 
   @Override
   public int hashCode() {
-    return Objects.hash(super.hashCode(), name, type);
+    return Objects.hash(super.hashCode(), name, type, project, labels);
   }
 }

core/src/main/java/feast/core/util/TypeConversion.java

@@ -70,9 +70,6 @@ public static Map<String, String> convertJsonStringToMap(String jsonString) {
    * @return json string corresponding to given map
    */
   public static String convertMapToJsonString(Map<String, String> map) {
-    if (map.isEmpty()) {
-      return "{}";
-    }
     return gson.toJson(map);
   }
 

core/src/main/java/feast/core/validators/FeatureSetValidator.java

@@ -27,13 +27,17 @@
 import java.util.stream.Collectors;
 
 public class FeatureSetValidator {
+
   public static void validateSpec(FeatureSet featureSet) {
     if (featureSet.getSpec().getProject().isEmpty()) {
       throw new IllegalArgumentException("Project name must be provided");
     }
     if (featureSet.getSpec().getName().isEmpty()) {
       throw new IllegalArgumentException("Feature set name must be provided");
     }
+    if (featureSet.getSpec().getLabelsMap().containsKey("")) {
+      throw new IllegalArgumentException("Feature set label keys must not be empty");
+    }
 
     checkValidCharacters(featureSet.getSpec().getProject(), "project");
     checkValidCharacters(featureSet.getSpec().getName(), "name");
@@ -44,6 +48,9 @@ public static void validateSpec(FeatureSet featureSet) {
     }
     for (FeatureSpec featureSpec : featureSet.getSpec().getFeaturesList()) {
       checkValidCharacters(featureSpec.getName(), "features::name");
+      if (featureSpec.getLabelsMap().containsKey("")) {
+        throw new IllegalArgumentException("Feature label keys must not be empty");
+      }
     }
   }
 

core/src/test/java/feast/core/service/JobServiceTest.java

@@ -34,11 +34,8 @@
 import feast.core.CoreServiceProto.RestartIngestionJobResponse;
 import feast.core.CoreServiceProto.StopIngestionJobRequest;
 import feast.core.CoreServiceProto.StopIngestionJobResponse;
-import feast.core.FeatureSetProto.FeatureSetStatus;
 import feast.core.FeatureSetReferenceProto.FeatureSetReference;
 import feast.core.IngestionJobProto.IngestionJob;
-import feast.core.SourceProto.KafkaSourceConfig;
-import feast.core.SourceProto.SourceType;
 import feast.core.StoreProto.Store.RedisConfig;
 import feast.core.StoreProto.Store.StoreType;
 import feast.core.dao.JobRepository;
@@ -84,14 +81,7 @@ public void setup() {
 
     // create mock objects for testing
     // fake data source
-    this.dataSource =
-        new Source(
-            SourceType.KAFKA,
-            KafkaSourceConfig.newBuilder()
-                .setBootstrapServers("kafka:9092")
-                .setTopic("my-topic")
-                .build(),
-            true);
+    this.dataSource = TestObjectFactory.defaultSource;
     // fake data store
     this.dataStore =
         new Store(
@@ -158,19 +148,12 @@ public void setupJobManager() {
 
   // dummy model constructorss
   private FeatureSet newDummyFeatureSet(String name, int version, String project) {
-    Field feature = new Field(name + "_feature", Enum.INT64);
-    Field entity = new Field(name + "_entity", Enum.STRING);
+    Field feature = TestObjectFactory.CreateFeatureField(name + "_feature", Enum.INT64);
+    Field entity = TestObjectFactory.CreateEntityField(name + "_entity", Enum.STRING);
 
     FeatureSet fs =
-        new FeatureSet(
-            name,
-            project,
-            version,
-            100L,
-            Arrays.asList(entity),
-            Arrays.asList(feature),
-            this.dataSource,
-            FeatureSetStatus.STATUS_READY);
+        TestObjectFactory.CreateFeatureSet(
+            name, project, version, Arrays.asList(entity), Arrays.asList(feature));
     fs.setCreated(Date.from(Instant.ofEpochSecond(10L)));
     return fs;
   }