Add maxage to feature set facade

zhilingc · zhilingc · commit edd0796d5ce8 · 2019-10-23T13:38:31.000+08:00
diff --git a/core/src/main/java/feast/core/model/FeatureSet.java b/core/src/main/java/feast/core/model/FeatureSet.java
@@ -1,5 +1,6 @@
 package feast.core.model;
 
+import com.google.protobuf.Duration;
 import com.google.protobuf.InvalidProtocolBufferException;
 import feast.core.FeatureSetProto.EntitySpec;
 import feast.core.FeatureSetProto.FeatureSetSpec;
@@ -43,6 +44,10 @@ public class FeatureSet extends AbstractTimestampEntity implements Comparable<Fe
   @Column(name = "version")
   private int version;
 
+  // Max allowed staleness for features in this featureSet.
+  @Column(name = "max_age")
+  private long maxAgeSeconds;
+
   @OneToMany(cascade = CascadeType.ALL, fetch = FetchType.EAGER)
   @Fetch(value = FetchMode.SUBSELECT)
   @JoinColumn(name = "entities")
@@ -63,25 +68,17 @@ public FeatureSet() {
     super();
   }
 
-  public FeatureSet(String name, int version, List<Field> entities, List<Field> features,
+  public FeatureSet(String name, int version, long maxAgeSeconds, List<Field> entities, List<Field> features,
       Source source) {
     this.id = String.format("%s:%s", name, version);
     this.name = name;
     this.version = version;
+    this.maxAgeSeconds = maxAgeSeconds;
     this.entities = entities;
     this.features = features;
     this.source = source;
   }
 
-  /**
-   * Updates the source of the featureset to point to the correct topic.
-   *
-   * @param topic topic to write this featureset to.
-   */
-  public void updateSourceTopic(String topic) throws InvalidProtocolBufferException {
-    source.setTopic(topic);
-  }
-
   public static FeatureSet fromProto(FeatureSetSpec featureSetSpec) {
     Source source = Source.fromProto(featureSetSpec.getSource());
     String id = String.format("%s:%d", featureSetSpec.getName(), featureSetSpec.getVersion());
@@ -100,6 +97,7 @@ public static FeatureSet fromProto(FeatureSetSpec featureSetSpec) {
 
     return new FeatureSet(featureSetSpec.getName(),
         featureSetSpec.getVersion(),
+        featureSetSpec.getMaxAge().getSeconds(),
         entities,
         features,
         source);
@@ -124,6 +122,7 @@ public FeatureSetSpec toProto() throws InvalidProtocolBufferException {
     return FeatureSetSpec.newBuilder()
         .setName(name)
         .setVersion(version)
+        .setMaxAge(Duration.newBuilder().setSeconds(maxAgeSeconds))
         .addAllEntities(entitySpecs)
         .addAllFeatures(featureSpecs)
         .setSource(source.toProto())
diff --git a/core/src/test/java/feast/core/service/SpecServiceTest.java b/core/src/test/java/feast/core/service/SpecServiceTest.java
@@ -365,7 +365,7 @@ private FeatureSet newDummyFeatureSet(String name, int version) {
             .build();
     Field feature = new Field(name, "feature", Enum.INT64);
     Field entity = new Field(name, "entity", Enum.STRING);
-    return new FeatureSet(name, version, Arrays.asList(entity), Arrays.asList(feature),
+    return new FeatureSet(name, version, 100L, Arrays.asList(entity), Arrays.asList(feature),
         new Source(
             SourceType.KAFKA, kafkaFeatureSourceOptions.toByteArray()));
   }

Original file line number	Diff line number	Diff line change
`@@ -365,7 +365,7 @@ private FeatureSet newDummyFeatureSet(String name, int version) {`
`365`	`365`	`.build();`
`366`	`366`	`Field feature = new Field(name, "feature", Enum.INT64);`
`367`	`367`	`Field entity = new Field(name, "entity", Enum.STRING);`
`368`		`- return new FeatureSet(name, version, Arrays.asList(entity), Arrays.asList(feature),`
	`368`	`+ return new FeatureSet(name, version, 100L, Arrays.asList(entity), Arrays.asList(feature),`
`369`	`369`	`new Source(`
`370`	`370`	`SourceType.KAFKA, kafkaFeatureSourceOptions.toByteArray()));`
`371`	`371`	`}`