Add storage interfaces, basic file structure (#529)

* Add storage interfaces, basic file structure

* Apply spotless, add comments

* Move parseResponse and isEmpty to response object

* Make changes to write interface to be more beam-like

* Pass feature specs to the retriever

* Pass feature specs to online retriever

* Add FeatureSetRequest

* Add mistakenly removed TestUtil

* Add mistakenly removed TestUtil

ingestion/src/test/java/feast/ingestion/transform/ValidateFeatureRowsTest.java

@@ -16,8 +16,6 @@
  */
 package feast.ingestion.transform;
 
-import static org.junit.Assert.*;
-
 import feast.core.FeatureSetProto.EntitySpec;
 import feast.core.FeatureSetProto.FeatureSet;
 import feast.core.FeatureSetProto.FeatureSetSpec;

pom.xml

@@ -33,6 +33,8 @@
         <module>core</module>
         <module>serving</module>
         <module>sdk/java</module>
+        <module>storage/api</module>
+        <module>storage/connectors</module>
     </modules>
 
     <properties>

storage/api/pom.xml

@@ -0,0 +1,72 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+  <parent>
+    <groupId>dev.feast</groupId>
+    <artifactId>feast-parent</artifactId>
+    <version>${revision}</version>
+    <relativePath>../..</relativePath>
+  </parent>
+
+  <modelVersion>4.0.0</modelVersion>
+  <artifactId>feast-storage-api</artifactId>
+
+  <name>Feast Storage API</name>
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-dependency-plugin</artifactId>
+        <configuration>
+          <!-- Required for generated code to compile; annotations are common false positive -->
+          <ignoredUnusedDeclaredDependencies>
+            javax.annotation
+          </ignoredUnusedDeclaredDependencies>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
+
+  <dependencies>
+
+    <dependency>
+      <groupId>dev.feast</groupId>
+      <artifactId>datatypes-java</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.beam</groupId>
+      <artifactId>beam-sdks-java-core</artifactId>
+      <version>${org.apache.beam.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>com.google.auto.value</groupId>
+      <artifactId>auto-value-annotations</artifactId>
+      <version>1.6.6</version>
+    </dependency>
+
+    <dependency>
+      <groupId>com.google.auto.value</groupId>
+      <artifactId>auto-value</artifactId>
+      <version>1.6.6</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-lang3</artifactId>
+      <version>3.9</version>
+    </dependency>
+
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <version>4.12</version>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+
+</project>

storage/api/src/main/java/feast/storage/api/retrieval/BatchRetriever.java

@@ -0,0 +1,37 @@
+/*
+ * 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.storage.api.retrieval;
+
+import feast.serving.ServingAPIProto;
+import java.util.List;
+
+/** Interface for implementing user defined retrieval functionality from Batch/historical stores. */
+public interface BatchRetriever {
+
+  /**
+   * Get all features corresponding to the provided batch features request.
+   *
+   * @param request {@link ServingAPIProto.GetBatchFeaturesRequest} containing requested features
+   *     and file containing entity columns.
+   * @param featureSetRequests List of {@link FeatureSetRequest} to feature references in the
+   *     request tied to that feature set.
+   * @return {@link ServingAPIProto.Job} if successful, contains the location of the results, else
+   *     contains the error to be returned to the user.
+   */
+  ServingAPIProto.Job getBatchFeatures(
+      ServingAPIProto.GetBatchFeaturesRequest request, List<FeatureSetRequest> featureSetRequests);
+}

storage/api/src/main/java/feast/storage/api/retrieval/FeatureSetRequest.java

@@ -0,0 +1,53 @@
+/*
+ * 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.storage.api.retrieval;
+
+import com.google.auto.value.AutoValue;
+import com.google.common.collect.ImmutableSet;
+import feast.core.FeatureSetProto.FeatureSetSpec;
+import feast.serving.ServingAPIProto.FeatureReference;
+import java.util.List;
+
+@AutoValue
+public abstract class FeatureSetRequest {
+  public abstract FeatureSetSpec getSpec();
+
+  public abstract ImmutableSet<FeatureReference> getFeatureReferences();
+
+  public static Builder newBuilder() {
+    return new AutoValue_FeatureSetRequest.Builder();
+  }
+
+  @AutoValue.Builder
+  public abstract static class Builder {
+    public abstract Builder setSpec(FeatureSetSpec spec);
+
+    abstract ImmutableSet.Builder<FeatureReference> featureReferencesBuilder();
+
+    public Builder addAllFeatureReferences(List<FeatureReference> featureReferenceList) {
+      featureReferencesBuilder().addAll(featureReferenceList);
+      return this;
+    }
+
+    public Builder addFeatureReference(FeatureReference featureReference) {
+      featureReferencesBuilder().add(featureReference);
+      return this;
+    }
+
+    public abstract FeatureSetRequest build();
+  }
+}

storage/api/src/main/java/feast/storage/api/retrieval/OnlineRetriever.java

@@ -0,0 +1,36 @@
+/*
+ * 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.storage.api.retrieval;
+
+import feast.serving.ServingAPIProto.GetOnlineFeaturesRequest.EntityRow;
+import feast.types.FeatureRowProto.FeatureRow;
+import java.util.List;
+
+/** Interface for implementing user defined retrieval functionality from Online stores. */
+public interface OnlineRetriever {
+
+  /**
+   * Get all values corresponding to the request.
+   *
+   * @param entityRows list of entity rows in the feature request
+   * @param featureSetRequests List of {@link FeatureSetRequest} to feature references in the
+   *     request tied to that feature set.
+   * @return list of {@link OnlineRetrieverResponse} for each entity row
+   */
+  List<List<FeatureRow>> getOnlineFeatures(
+      List<EntityRow> entityRows, List<FeatureSetRequest> featureSetRequests);
+}

storage/api/src/main/java/feast/storage/api/retrieval/OnlineRetrieverResponse.java

@@ -0,0 +1,44 @@
+/*
+ * 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.storage.api.retrieval;
+
+import feast.types.FeatureRowProto;
+
+/** Response from an online store. */
+public interface OnlineRetrieverResponse {
+
+  /**
+   * Checks whether the response is empty, i.e. feature does not exist in the store
+   *
+   * @return boolean
+   */
+  boolean isEmpty();
+
+  /**
+   * Get the featureset associated with this response.
+   *
+   * @return String featureset reference in format featureSet:version
+   */
+  String getFeatureSet();
+
+  /**
+   * Parse response to FeatureRow
+   *
+   * @return {@link FeatureRowProto.FeatureRow}
+   */
+  FeatureRowProto.FeatureRow toFeatureRow();
+}

storage/api/src/main/java/feast/storage/api/write/DeadletterSink.java

@@ -0,0 +1,38 @@
+/*
+ * 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.storage.api.write;
+
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PDone;
+
+/** Interface for for implementing user defined deadletter sinks to write failed elements to. */
+public interface DeadletterSink {
+
+  /**
+   * Set up the deadletter sink for writes. This method will be called once during pipeline
+   * initialisation.
+   */
+  void prepareWrite();
+
+  /**
+   * Get a {@link PTransform} that writes a collection of FailedElements to the deadletter sink.
+   *
+   * @return {@link PTransform}
+   */
+  PTransform<PCollection<FailedElement>, PDone> write();
+}

storage/api/src/main/java/feast/storage/api/write/FailedElement.java

@@ -0,0 +1,83 @@
+/*
+ * 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.storage.api.write;
+
+import com.google.auto.value.AutoValue;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
+import org.joda.time.Instant;
+
+@AutoValue
+// Use DefaultSchema annotation so this AutoValue class can be serialized by Beam
+// https://issues.apache.org/jira/browse/BEAM-1891
+// https://github.com/apache/beam/pull/7334
+@DefaultSchema(AutoValueSchema.class)
+public abstract class FailedElement {
+  public abstract Instant getTimestamp();
+
+  @Nullable
+  public abstract String getJobName();
+
+  @Nullable
+  public abstract String getProjectName();
+
+  @Nullable
+  public abstract String getFeatureSetName();
+
+  @Nullable
+  public abstract String getFeatureSetVersion();
+
+  @Nullable
+  public abstract String getTransformName();
+
+  @Nullable
+  public abstract String getPayload();
+
+  @Nullable
+  public abstract String getErrorMessage();
+
+  @Nullable
+  public abstract String getStackTrace();
+
+  public static Builder newBuilder() {
+    return new AutoValue_FailedElement.Builder().setTimestamp(Instant.now());
+  }
+
+  @AutoValue.Builder
+  public abstract static class Builder {
+    public abstract Builder setTimestamp(Instant timestamp);
+
+    public abstract Builder setProjectName(String projectName);
+
+    public abstract Builder setFeatureSetName(String featureSetName);
+
+    public abstract Builder setFeatureSetVersion(String featureSetVersion);
+
+    public abstract Builder setJobName(String jobName);
+
+    public abstract Builder setTransformName(String transformName);
+
+    public abstract Builder setPayload(String payload);
+
+    public abstract Builder setErrorMessage(String errorMessage);
+
+    public abstract Builder setStackTrace(String stackTrace);
+
+    public abstract FailedElement build();
+  }
+}