local read-write path

Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com>

docs/specs/online_store_format.md

@@ -59,20 +59,20 @@ 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
@@ -90,7 +90,7 @@ Other types of entity keys are not supported in this version of the specificatio
 
 **Example:**
 
-![Firestore Online Example](firebase_online_example.png)
+![Datastore Online Example](datastore_online_example.png)
 
 # Appendix
 

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)