Read and write path for Datastore and SQLite

docs/specs/online_store_format.md

@@ -59,38 +59,38 @@ Here's an example of how the entire thing looks like:
 
 However, we'll address this issue in future versions of the protocol.
 
-## Cloud Firestore Online Store Format
+## Google Datastore Online Store Format
 
-[Firebase data model](https://firebase.google.com/docs/firestore/data-model) is a hierarchy of documents that can contain (sub)-collections. This structure can be multiple levels deep; documents and subcollections are alternating in this hierarchy.
+[Datastore data model](https://cloud.google.com/datastore/docs/concepts/entities) is a collection of documents called Entities (not to be confused with Feast Entities). Documents can be organized in a hierarchy using Kinds.
 
-We use the following structure to store feature data in the Firestore:
-* at the first level, there is a collection for each Feast project
-* second level, in each project-collection, there is a Firebase document for each Feature Table
-* third level, in the document for the Feature Table, there is a subcollection called `values` that contain a document per feature row. That document contains the following fields:
-  * `key` contains entity key as serialized `feast.types.EntityKey` proto
-  * `values` contains feature name to value map, values serialized as `feast.types.Value` proto
-  * `event_ts` contains event timestamp (in the native firestore timestamp format)
-  * `created_ts` contains write timestamp (in the native firestore timestamp format)
+We use the following structure to store feature data in Datastore:
+* There is a Datastore Entity for each Feast Project, with Kind `Project`.
+* Under that Datastore Entity, there is a Datastore Entity for each Feast Feature Table or View, with Kind `Table`. That contains one additional field, `created_ts` that contains the timestamp when this Datastore Entity was created.
+* Under that Datastore Entity, there is a Datastore Entity for each Feast Entity Key with Kind `Row`. That contains the following fields:
+	 * `key` contains entity key as serialized `feast.types.EntityKey` proto
+	 * `values` contains feature name to value map, values serialized as `feast.types.Value` proto
+   * `event_ts` contains event timestamp (in the datastore timestamp format)
+   * `created_ts` contains write timestamp (in the datastore timestamp format).
 
-Document id for the feature document is computed by hashing entity key using murmurhash3_128 algorithm as follows:
+The id for the `Row` Datastore Entity is computed by hashing entity key using murmurhash3_128 algorithm as follows:
 
-1. hash entity names, sorted in alphanumeric order, by serializing them to bytes using the Value Serialization steps below
-2. hash the entity values in the same order as corresponding entity names, by serializing them to bytes using the Value Serialization steps below
+1. Hash entity names, sorted in alphanumeric order, by serializing them to bytes using the Value Serialization steps below.
+2. Hash the entity values in the same order as corresponding entity names, by serializing them to bytes using the Value Serialization steps below.
 
 Value Serialization:
 * Store the type of the value (ValueType enum) as little-endian uint32.
-* Store the byte length of the serialized value as little-endian uint32
+* Store the byte length of the serialized value as little-endian uint32.
 * Store the serialized value as bytes:
   - binary values are serialized as is
   - string values serialized as utf8 string
   - int64 and int32 hashed as little-endian byte representation (8 and 4 bytes respectively)
-  - bool hashed as 0 or 1 byte
+  - bool hashed as 0 or 1 byte.
 
 Other types of entity keys are not supported in this version of the specification, when using Cloud Firestore.
 
 **Example:**
 
-![Firestore Online Example](firebase_online_example.png)
+![Datastore Online Example](datastore_online_example.png)
 
 # Appendix
 

infra/docker/ci/Dockerfile

@@ -29,7 +29,7 @@ ENV MAVEN_HOME /usr/share/maven
 ENV MAVEN_CONFIG "/root/.m2"
 
 # Install Make and Python
-ENV PYTHON_VERSION 3.6
+ENV PYTHON_VERSION 3.7
 
 RUN apt-get install -y build-essential curl python${PYTHON_VERSION} \
     python${PYTHON_VERSION}-dev python${PYTHON_VERSION}-distutils && \
@@ -77,7 +77,7 @@ RUN PROTOC_ZIP=protoc-${PROTOC_VERSION}-linux-x86_64.zip && \
 RUN curl -sL https://aka.ms/InstallAzureCLIDeb | bash
 
 # Install kubectl
-RUN apt-get install -y kubectl=1.20.2-00
+RUN apt-get install -y kubectl=1.20.4-00
 
 # Install helm
 RUN curl -fsSL -o get_helm.sh https://raw.githubusercontent.com/helm/helm/master/scripts/get-helm-3 && \

sdk/python/feast/feature_store.py

@@ -11,24 +11,47 @@
 # 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.
+from pathlib import Path
 from typing import Optional
 
-from feast.repo_config import RepoConfig, load_repo_config
+from feast.infra.provider import Provider, get_provider
+from feast.registry import Registry
+from feast.repo_config import (
+    LocalOnlineStoreConfig,
+    OnlineStoreConfig,
+    RepoConfig,
+    load_repo_config,
+)
 
 
 class FeatureStore:
     """
     A FeatureStore object is used to define, create, and retrieve features.
     """
 
+    config: RepoConfig
+
     def __init__(
-        self, config_path: Optional[str], config: Optional[RepoConfig],
+        self, repo_path: Optional[str], config: Optional[RepoConfig],
     ):
-        if config_path is None or config is None:
-            raise Exception("You cannot specify both config_path and config")
+        if repo_path is not None and config is not None:
+            raise Exception("You cannot specify both repo_path and config")
         if config is not None:
             self.config = config
-        elif config_path is not None:
-            self.config = load_repo_config(config_path)
+        elif repo_path is not None:
+            self.config = load_repo_config(Path(repo_path))
         else:
-            self.config = RepoConfig()
+            self.config = RepoConfig(
+                metadata_store="./metadata.db",
+                project="default",
+                provider="local",
+                online_store=OnlineStoreConfig(
+                    local=LocalOnlineStoreConfig("online_store.db")
+                ),
+            )
+
+    def _get_provider(self) -> Provider:
+        return get_provider(self.config)
+
+    def _get_registry(self) -> Registry:
+        return Registry(self.config.metadata_store)

sdk/python/feast/infra/gcp.py

@@ -1,17 +1,24 @@
 from datetime import datetime
-from typing import List, Optional
+from typing import Dict, List, Optional, Tuple
+
+import mmh3
+from pytz import utc
 
 from feast import FeatureTable
 from feast.infra.provider import Provider
 from feast.repo_config import DatastoreOnlineStoreConfig
+from feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto
+from feast.types.Value_pb2 import Value as ValueProto
+
+from .key_encoding_utils import serialize_entity_key
 
 
 def _delete_all_values(client, key) -> None:
     """
     Delete all data under the key path in datastore.
     """
     while True:
-        query = client.query(kind="Value", ancestor=key)
+        query = client.query(kind="Row", ancestor=key)
         entities = list(query.fetch(limit=1000))
         if not entities:
             return
@@ -21,19 +28,37 @@ def _delete_all_values(client, key) -> None:
             client.delete(entity.key)
 
 
+def compute_datastore_entity_id(entity_key: EntityKeyProto) -> str:
+    """
+    Compute Datastore Entity id given Feast Entity Key.
+
+    Remember that Datastore Entity is a concept from the Datastore data model, that has nothing to
+    do with the Entity concept we have in Feast.
+    """
+    return mmh3.hash_bytes(serialize_entity_key(entity_key)).hex()
+
+
+def _make_tzaware(t: datetime):
+    """ We assume tz-naive datetimes are UTC """
+    if t.tzinfo is None:
+        return t.replace(tzinfo=utc)
+    else:
+        return t
+
+
 class Gcp(Provider):
-    _project_id: Optional[str]
+    _gcp_project_id: Optional[str]
 
     def __init__(self, config: Optional[DatastoreOnlineStoreConfig]):
         if config:
-            self._project_id = config.project_id
+            self._gcp_project_id = config.project_id
         else:
-            self._project_id = None
+            self._gcp_project_id = None
 
     def _initialize_client(self):
         from google.cloud import datastore
 
-        if self._project_id is not None:
+        if self._gcp_project_id is not None:
             return datastore.Client(self.project_id)
         else:
             return datastore.Client()
@@ -49,28 +74,88 @@ def update_infra(
         client = self._initialize_client()
 
         for table in tables_to_keep:
-            key = client.key("FeastProject", project, "FeatureTable", table.name)
+            key = client.key("Project", project, "Table", table.name)
             entity = datastore.Entity(key=key)
-            entity.update({"created_at": datetime.utcnow()})
+            entity.update({"created_ts": datetime.utcnow()})
             client.put(entity)
 
         for table in tables_to_delete:
             _delete_all_values(
-                client, client.key("FeastProject", project, "FeatureTable", table.name)
+                client, client.key("Project", project, "Table", table.name)
             )
 
             # Delete the table metadata datastore entity
-            key = client.key("FeastProject", project, "FeatureTable", table.name)
+            key = client.key("Project", project, "Table", table.name)
             client.delete(key)
 
     def teardown_infra(self, project: str, tables: List[FeatureTable]) -> None:
         client = self._initialize_client()
 
         for table in tables:
             _delete_all_values(
-                client, client.key("FeastProject", project, "FeatureTable", table.name)
+                client, client.key("Project", project, "Table", table.name)
             )
 
             # Delete the table metadata datastore entity
-            key = client.key("FeastProject", project, "FeatureTable", table.name)
+            key = client.key("Project", project, "Table", table.name)
             client.delete(key)
+
+    def online_write_batch(
+        self,
+        project: str,
+        table: FeatureTable,
+        data: List[Tuple[EntityKeyProto, Dict[str, ValueProto], datetime]],
+        created_ts: datetime,
+    ) -> None:
+        from google.cloud import datastore
+
+        client = self._initialize_client()
+
+        for entity_key, features, timestamp in data:
+            document_id = compute_datastore_entity_id(entity_key)
+
+            key = client.key(
+                "Project", project, "Table", table.name, "Row", document_id,
+            )
+            with client.transaction():
+                entity = client.get(key)
+                if entity is not None:
+                    if entity["event_ts"] > _make_tzaware(timestamp):
+                        # Do not overwrite feature values computed from fresher data
+                        continue
+                    elif entity["event_ts"] == _make_tzaware(timestamp) and entity[
+                        "created_ts"
+                    ] > _make_tzaware(created_ts):
+                        # Do not overwrite feature values computed from the same data, but
+                        # computed later than this one
+                        continue
+                else:
+                    entity = datastore.Entity(key=key)
+
+                entity.update(
+                    dict(
+                        key=entity_key.SerializeToString(),
+                        values={k: v.SerializeToString() for k, v in features.items()},
+                        event_ts=_make_tzaware(timestamp),
+                        created_ts=_make_tzaware(created_ts),
+                    )
+                )
+                client.put(entity)
+
+    def online_read(
+        self, project: str, table: FeatureTable, entity_key: EntityKeyProto
+    ) -> Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]:
+        client = self._initialize_client()
+
+        document_id = compute_datastore_entity_id(entity_key)
+        key = client.key("Project", project, "Table", table.name, "Row", document_id)
+        value = client.get(key)
+        if value is not None:
+            res = {}
+            for feature_name, value_bin in value["values"].items():
+                val = ValueProto()
+                val.ParseFromString(value_bin)
+                res[feature_name] = val
+            return value["event_ts"], res
+        else:
+            return None, None

sdk/python/feast/infra/key_encoding_utils.py

@@ -0,0 +1,48 @@
+import struct
+from typing import List, Tuple
+
+from feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto
+from feast.types.Value_pb2 import Value as ValueProto
+from feast.types.Value_pb2 import ValueType
+
+
+def _serialize_val(value_type, v: ValueProto) -> Tuple[bytes, int]:
+    if value_type == "string_val":
+        return v.string_val.encode("utf8"), ValueType.STRING
+    elif value_type == "bytes_val":
+        return v.bytes_val, ValueType.BYTES
+    elif value_type == "int32_val":
+        return struct.pack("<i", v.int32_val), ValueType.INT32
+    elif value_type == "int64_val":
+        return struct.pack("<l", v.int64_val), ValueType.INT64
+    else:
+        raise ValueError(f"Value type not supported for Firestore: {v}")
+
+
+def serialize_entity_key(entity_key: EntityKeyProto) -> bytes:
+    """
+    Serialize entity key to a bytestring so it can be used as a lookup key in a hash table.
+
+    We need this encoding to be stable; therefore we cannot just use protobuf serialization
+    here since it does not guarantee that two proto messages containing the same data will
+    serialize to the same byte string[1].
+
+    [1] https://developers.google.com/protocol-buffers/docs/encoding
+    """
+    sorted_keys, sorted_values = zip(
+        *sorted(zip(entity_key.entity_names, entity_key.entity_values))
+    )
+
+    output: List[bytes] = []
+    for k in sorted_keys:
+        output.append(struct.pack("<I", ValueType.STRING))
+        output.append(k.encode("utf8"))
+    for v in sorted_values:
+        val_bytes, value_type = _serialize_val(v.WhichOneof("val"), v)
+
+        output.append(struct.pack("<I", value_type))
+
+        output.append(struct.pack("<I", len(val_bytes)))
+        output.append(val_bytes)
+
+    return b"".join(output)