Checkpoint

davidheryanto · davidheryanto · commit e14f89b3b991 · 2019-09-30T18:09:02.000+08:00
diff --git a/serving/src/main/java/feast/serving/configuration/ServingServiceConfig.java b/serving/src/main/java/feast/serving/configuration/ServingServiceConfig.java
@@ -69,7 +69,7 @@ public ServingService servingService(
         BigQueryConfig bqConfig = store.getBigqueryConfig();
         BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();
         servingService =
-            new BigQueryServingService(bigquery, bqConfig.getProjectId(), bqConfig.getDatasetId());
+            new BigQueryServingService(bigquery, bqConfig.getProjectId(), bqConfig.getDatasetId(), specService);
         break;
       case CASSANDRA:
       case UNRECOGNIZED:
diff --git a/serving/src/main/java/feast/serving/service/BigQueryServingService.java b/serving/src/main/java/feast/serving/service/BigQueryServingService.java
@@ -1,6 +1,9 @@
 package feast.serving.service;
 
 import com.google.cloud.bigquery.BigQuery;
+import feast.core.CoreServiceProto.GetFeatureSetsRequest;
+import feast.core.CoreServiceProto.GetFeatureSetsRequest.Filter;
+import feast.core.FeatureSetProto.FeatureSetSpec;
 import feast.serving.ServingAPIProto.FeastServingType;
 import feast.serving.ServingAPIProto.GetBatchFeaturesFromCompletedJobRequest;
 import feast.serving.ServingAPIProto.GetBatchFeaturesFromCompletedJobResponse;
@@ -15,17 +18,23 @@
 import feast.serving.ServingAPIProto.LoadBatchFeaturesResponse;
 import feast.serving.ServingAPIProto.ReloadJobStatusRequest;
 import feast.serving.ServingAPIProto.ReloadJobStatusResponse;
+import feast.serving.util.BigQueryUtil;
 import io.grpc.Status;
+import java.util.List;
+import java.util.stream.Collectors;
 
 public class BigQueryServingService implements ServingService {
   private final BigQuery bigquery;
   private final String projectId;
   private final String datasetId;
+  private final SpecService specService;
 
-  public BigQueryServingService(BigQuery bigquery, String projectId, String datasetId) {
+  public BigQueryServingService(
+      BigQuery bigquery, String projectId, String datasetId, SpecService specService) {
     this.bigquery = bigquery;
     this.projectId = projectId;
     this.datasetId = datasetId;
+    this.specService = specService;
   }
 
   @Override
@@ -43,7 +52,26 @@ public GetOnlineFeaturesResponse getOnlineFeatures(GetFeaturesRequest getFeature
 
   @Override
   public GetBatchFeaturesResponse getBatchFeatures(GetFeaturesRequest getFeaturesRequest) {
-    return null;
+    List<FeatureSetSpec> featureSetSpecs =
+        getFeaturesRequest.getFeatureSetsList().stream()
+            .map(
+                featureSet ->
+                    specService.getFeatureSets(
+                        GetFeatureSetsRequest.newBuilder()
+                            .setFilter(
+                                Filter.newBuilder().setFeatureSetName(featureSet.getName()).build())
+                            .build()))
+            .map(response -> response.getFeatureSetsList().get(0))
+            .collect(Collectors.toList());
+    String query =
+        BigQueryUtil.createQuery(
+            getFeaturesRequest.getFeatureSetsList(),
+            featureSetSpecs,
+            getFeaturesRequest.getEntityDataset().getEntityNamesList(),
+            getFeaturesRequest.getEntityDataset().getEntityDatasetRowsList(),
+            datasetId);
+    System.out.println(query);
+    return GetBatchFeaturesResponse.getDefaultInstance();
   }
 
   @Override
diff --git a/serving/src/main/java/feast/serving/service/JobService.java b/serving/src/main/java/feast/serving/service/JobService.java
@@ -0,0 +1,24 @@
+package feast.serving.service;
+
+import feast.serving.ServingAPIProto.Job;
+import java.util.Optional;
+
+// JobService interface specifies the operations to manage Job instances internally in Feast
+
+public interface JobService {
+
+  /**
+   * Get Job by job id.
+   *
+   * @param id job id
+   * @return feast.serving.ServingAPIProto.Job
+   */
+  public Optional<Job> get(String id);
+
+  /**
+   * Update or create a job (if not exists)
+   *
+   * @param job feast.serving.ServingAPIProto.Job
+   */
+  public void upsert(Job job);
+}
diff --git a/serving/src/main/java/feast/serving/service/RedisBackedJobService.java b/serving/src/main/java/feast/serving/service/RedisBackedJobService.java
@@ -0,0 +1,47 @@
+package feast.serving.service;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.util.JsonFormat;
+import feast.serving.ServingAPIProto.Job;
+import feast.serving.ServingAPIProto.Job.Builder;
+import java.util.Optional;
+import lombok.extern.slf4j.Slf4j;
+import redis.clients.jedis.JedisPool;
+
+@Slf4j
+public class RedisBackedJobService implements JobService {
+  private final JedisPool jedisPool;
+
+  public RedisBackedJobService(JedisPool jedisPool) {
+    this.jedisPool = jedisPool;
+  }
+
+  @Override
+  public Optional<Job> get(String id) {
+    String json = jedisPool.getResource().get(id);
+    if (json == null) {
+      return Optional.empty();
+    }
+    Job job = null;
+    Builder builder = Job.newBuilder();
+    try {
+      JsonFormat.parser().merge(json, builder);
+      job = builder.build();
+    } catch (InvalidProtocolBufferException e) {
+      log.error(String.format("Failed to parse JSON for Feast job: %s", e.getMessage()));
+    }
+
+    return Optional.ofNullable(job);
+  }
+
+  @Override
+  public void upsert(Job job) {
+    try {
+      jedisPool
+          .getResource()
+          .set(job.getId(), JsonFormat.printer().omittingInsignificantWhitespace().print(job));
+    } catch (InvalidProtocolBufferException e) {
+      log.error(String.format("Failed to upsert job: %s", e.getMessage()));
+    }
+  }
+}
