Add computation and retrieval of batch feature statistics

core/pom.xml

@@ -57,6 +57,24 @@
             <version>${project.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>dev.feast</groupId>
+            <artifactId>feast-storage-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>dev.feast</groupId>
+            <artifactId>feast-storage-connector-bigquery</artifactId>
+            <version>${project.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.beam</groupId>
+                    <artifactId>*</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
         <!-- Hot reloading for Spring Boot. spring-boot-maven-plugin removes
              this automatically when packaging. -->
         <dependency>

core/src/main/java/feast/core/dao/MetricsRepository.java → core/src/main/java/feast/core/dao/FeatureStatisticsRepository.java

@@ -16,12 +16,16 @@
  */
 package feast.core.dao;
 
-import feast.core.model.Metrics;
-import java.util.List;
+import feast.core.model.Feature;
+import feast.core.model.FeatureStatistics;
+import java.util.Date;
+import java.util.Optional;
 import org.springframework.data.jpa.repository.JpaRepository;
-import org.springframework.stereotype.Repository;
 
-@Repository
-public interface MetricsRepository extends JpaRepository<Metrics, Long> {
-  List<Metrics> findByJob_Id(String id);
+/** JPA repository supplying Statistics objects for features keyed by id. */
+public interface FeatureStatisticsRepository extends JpaRepository<FeatureStatistics, Integer> {
+  Optional<FeatureStatistics> findFeatureStatisticsByFeatureAndDatasetId(
+      Feature feature, String datasetId);
+
+  Optional<FeatureStatistics> findFeatureStatisticsByFeatureAndDate(Feature feature, Date date);
 }

core/src/main/java/feast/core/grpc/CoreServiceImpl.java

@@ -19,36 +19,14 @@
 import com.google.api.gax.rpc.InvalidArgumentException;
 import com.google.protobuf.InvalidProtocolBufferException;
 import feast.core.CoreServiceGrpc.CoreServiceImplBase;
-import feast.core.CoreServiceProto.ApplyFeatureSetRequest;
-import feast.core.CoreServiceProto.ApplyFeatureSetResponse;
-import feast.core.CoreServiceProto.ArchiveProjectRequest;
-import feast.core.CoreServiceProto.ArchiveProjectResponse;
-import feast.core.CoreServiceProto.CreateProjectRequest;
-import feast.core.CoreServiceProto.CreateProjectResponse;
-import feast.core.CoreServiceProto.GetFeastCoreVersionRequest;
-import feast.core.CoreServiceProto.GetFeastCoreVersionResponse;
-import feast.core.CoreServiceProto.GetFeatureSetRequest;
-import feast.core.CoreServiceProto.GetFeatureSetResponse;
-import feast.core.CoreServiceProto.ListFeatureSetsRequest;
-import feast.core.CoreServiceProto.ListFeatureSetsResponse;
-import feast.core.CoreServiceProto.ListIngestionJobsRequest;
-import feast.core.CoreServiceProto.ListIngestionJobsResponse;
-import feast.core.CoreServiceProto.ListProjectsRequest;
-import feast.core.CoreServiceProto.ListProjectsResponse;
-import feast.core.CoreServiceProto.ListStoresRequest;
-import feast.core.CoreServiceProto.ListStoresResponse;
-import feast.core.CoreServiceProto.RestartIngestionJobRequest;
-import feast.core.CoreServiceProto.RestartIngestionJobResponse;
-import feast.core.CoreServiceProto.StopIngestionJobRequest;
-import feast.core.CoreServiceProto.StopIngestionJobResponse;
-import feast.core.CoreServiceProto.UpdateStoreRequest;
-import feast.core.CoreServiceProto.UpdateStoreResponse;
+import feast.core.CoreServiceProto.*;
 import feast.core.exception.RetrievalException;
 import feast.core.grpc.interceptors.MonitoringInterceptor;
 import feast.core.model.Project;
 import feast.core.service.AccessManagementService;
 import feast.core.service.JobService;
 import feast.core.service.SpecService;
+import feast.core.service.StatsService;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import io.grpc.stub.StreamObserver;
@@ -65,15 +43,18 @@
 public class CoreServiceImpl extends CoreServiceImplBase {
 
   private SpecService specService;
+  private StatsService statsService;
   private AccessManagementService accessManagementService;
   private JobService jobService;
 
   @Autowired
   public CoreServiceImpl(
       SpecService specService,
+      StatsService statsService,
       AccessManagementService accessManagementService,
       JobService jobService) {
     this.specService = specService;
+    this.statsService = statsService;
     this.accessManagementService = accessManagementService;
     this.jobService = jobService;
   }
@@ -113,6 +94,32 @@ public void listFeatureSets(
     }
   }
 
+  @Override
+  public void getFeatureStatistics(
+      GetFeatureStatisticsRequest request,
+      StreamObserver<GetFeatureStatisticsResponse> responseObserver) {
+    try {
+      GetFeatureStatisticsResponse response = statsService.getFeatureStatistics(request);
+      responseObserver.onNext(response);
+      responseObserver.onCompleted();
+    } catch (IllegalArgumentException e) {
+      log.error("Illegal arguments provided to GetFeatureStatistics method: ", e);
+      responseObserver.onError(
+          Status.INVALID_ARGUMENT
+              .withDescription(e.getMessage())
+              .withCause(e)
+              .asRuntimeException());
+    } catch (RetrievalException e) {
+      log.error("Unable to fetch feature set requested in GetFeatureStatistics method: ", e);
+      responseObserver.onError(
+          Status.NOT_FOUND.withDescription(e.getMessage()).withCause(e).asRuntimeException());
+    } catch (Exception e) {
+      log.error("Exception has occurred in GetFeatureStatistics method: ", e);
+      responseObserver.onError(
+          Status.INTERNAL.withDescription(e.getMessage()).withCause(e).asRuntimeException());
+    }
+  }
+
   @Override
   public void listStores(
       ListStoresRequest request, StreamObserver<ListStoresResponse> responseObserver) {

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

@@ -0,0 +1,69 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2018-2019 The Feast Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package feast.core.model;
+
+import feast.core.FeatureSetProto.EntitySpec;
+import feast.types.ValueProto.ValueType;
+import java.util.Objects;
+import javax.persistence.EmbeddedId;
+import lombok.Getter;
+import lombok.Setter;
+
+/** Feast entity object. Contains name, type as well as domain metadata about the entity. */
+@Getter
+@Setter
+@javax.persistence.Entity
+public class Entity {
+  @EmbeddedId private EntityReference reference;
+
+  private String type;
+
+  public Entity() {}
+
+  private Entity(String name, ValueType.Enum type) {
+    this.setReference(new EntityReference(name));
+    this.setType(type.toString());
+  }
+
+  public static Entity withRef(EntityReference entityRef) {
+    Entity entity = new Entity();
+    entity.setReference(entityRef);
+    return entity;
+  }
+
+  public static Entity fromProto(EntitySpec entitySpec) {
+    Entity entity = new Entity(entitySpec.getName(), entitySpec.getValueType());
+    return entity;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    Entity feature = (Entity) o;
+    return getReference().equals(feature.getReference()) && getType().equals(feature.getType());
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(super.hashCode(), getReference(), getType());
+  }
+}

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

@@ -0,0 +1,72 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2018-2020 The Feast Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package feast.core.model;
+
+import java.io.Serializable;
+import java.util.Objects;
+import javax.persistence.Column;
+import javax.persistence.Embeddable;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+@Embeddable
+@NoArgsConstructor
+@AllArgsConstructor
+@Getter
+@Setter
+public class EntityReference implements Serializable {
+  // Project the field belongs to
+  @Column(nullable = false)
+  private String project;
+
+  // Feature set the field belongs to
+  @Column(name = "feature_set", nullable = false)
+  private String featureSet;
+
+  // Version of the feature set this field belongs to
+  @Column(nullable = false)
+  private int version;
+
+  // Name of the field
+  @Column(nullable = false)
+  private String name;
+
+  EntityReference(String name) {
+    this.name = name;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    EntityReference fieldId = (EntityReference) o;
+    return Objects.equals(name, fieldId.getName())
+        && Objects.equals(project, fieldId.getProject())
+        && Objects.equals(featureSet, fieldId.getFeatureSet());
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(super.hashCode(), project, featureSet, name);
+  }
+}