From 3d88358af70657f2a305b5f3618817ce9aa25594 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 19 May 2022 10:22:30 -0700 Subject: [PATCH 01/17] feat: WIP SQLAlchemy Registry Support Signed-off-by: Achal Shah --- sdk/python/feast/infra/registry_stores/sql.py | 211 ++++++++++++++++++ sdk/python/feast/repo_config.py | 4 + setup.py | 1 + 3 files changed, 216 insertions(+) create mode 100644 sdk/python/feast/infra/registry_stores/sql.py diff --git a/sdk/python/feast/infra/registry_stores/sql.py b/sdk/python/feast/infra/registry_stores/sql.py new file mode 100644 index 00000000000..0feee8e5ab4 --- /dev/null +++ b/sdk/python/feast/infra/registry_stores/sql.py @@ -0,0 +1,211 @@ +from datetime import datetime +from pathlib import Path +from typing import List, Optional + +from sql import ( # type: ignore + BIGINT, + VARBINARY, + Column, + MetaData, + String, + Table, + create_engine, + delete, + insert, + select, + update, +) +from sql.engine import Engine + +from feast.data_source import DataSource +from feast.entity import Entity +from feast.errors import DataSourceObjectNotFoundException, EntityNotFoundException +from feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto +from feast.protos.feast.core.Entity_pb2 import Entity as EntityProto +from feast.registry import Registry +from feast.repo_config import RegistryConfig + +metadata = MetaData() + +entities = Table( + "entities", + metadata, + Column("entity_id", String, primary_key=True), + Column("last_updated_timestamp", BIGINT, nullable=False), + Column("entity_proto", VARBINARY, nullable=False), +) + +data_sources = Table( + "data_sources", + metadata, + Column("data_source_name", String, primary_key=True), + Column("last_updated_timestamp", BIGINT, nullable=False), + Column("data_source_proto", VARBINARY, nullable=False), +) + +feature_views = Table( + "feature_views", + metadata, + Column("feature_view_name", String, primary_key=True), + Column("last_updated_timestamp", BIGINT, nullable=False), + Column("materialized_intervals", VARBINARY, nullable=False), + Column("feature_view_proto", VARBINARY, nullable=False), +) + +request_feature_views = Table( + "request_feature_views", + metadata, + Column("feature_view_name", String, primary_key=True), + Column("last_updated_timestamp", BIGINT, nullable=False), + Column("feature_view_proto", VARBINARY, nullable=False), +) + +on_demand_feature_views = Table( + "on_demand_feature_views", + metadata, + Column("feature_view_name", String, primary_key=True), + Column("last_updated_timestamp", BIGINT, nullable=False), + Column("feature_view_proto", VARBINARY, nullable=False), +) + +feature_user_metadata = Table( + "feature_metadata", + metadata, + Column("feature_name", String, primary_key=True), + Column("last_updated_timestamp", BIGINT, nullable=False), + Column("feature_metadata_binary", VARBINARY, nullable=False), +) + +feature_services = Table( + "feature_services", + metadata, + Column("feature_service_name", String, primary_key=True), + Column("last_updated_timestamp", BIGINT, nullable=False), + Column("feature_service_proto", VARBINARY, nullable=False), +) + +APPLY_OPERATIONS = {"entity": (entities, entities.c.entity_id)} + + +class SqlRegistry(Registry): + def __init__( + self, registry_config: Optional[RegistryConfig], repo_path: Optional[Path] + ): + assert registry_config + self.engine: Engine = create_engine(registry_config.path, echo=True) + metadata.create_all(self.engine) + + def teardown(self): + super().teardown() + + def apply_entity(self, entity: Entity, project: str, commit: bool = True): + with self.engine.connect() as conn: + stmt = select(entities).where(entities.c.entity_id == entity.name) + entity.last_updated_timestamp = datetime.utcnow() + row = conn.execute(stmt).first() + if row: + update_stmt = ( + update(entities) + .where(entities.c.entity_id == entity.name,) + .values( + entity_proto=entity.to_proto().SerializeToString(), + last_updated_timestamp=int( + entity.last_updated_timestamp.timestamp() + ), + ) + ) + conn.execute(update_stmt) + else: + insert_stmt = insert(entities).values( + entity_id=entity.name, + entity_proto=entity.to_proto().SerializeToString(), + last_updated_timestamp=int( + entity.last_updated_timestamp.timestamp() + ), + ) + conn.execute(insert_stmt) + + def get_entity(self, name: str, project: str, allow_cache: bool = False) -> Entity: + with self.engine.connect() as conn: + stmt = select(entities).where(entities.c.entity_id == name) + row = conn.execute(stmt).first() + if row: + entity_proto = EntityProto.FromString(row["entity_proto"]) + return Entity.from_proto(entity_proto) + raise EntityNotFoundException(name, project=project) + + def list_entities(self, project: str, allow_cache: bool = False) -> List[Entity]: + with self.engine.connect() as conn: + stmt = select(entities) + rows = conn.execute(stmt).all() + if rows: + return [ + Entity.from_proto(EntityProto.FromString(row["entity_proto"])) + for row in rows + ] + return [] + + def delete_entity(self, name: str, project: str, commit: bool = True): + with self.engine.connect() as conn: + stmt = delete(entities).where(entities.c.entity_id == name) + rows = conn.execute(stmt) + if rows.rowcount < 1: + raise EntityNotFoundException(name, project) + + def get_data_source( + self, name: str, project: str, allow_cache: bool = False + ) -> DataSource: + with self.engine.connect() as conn: + stmt = select(data_sources).where(data_sources.c.entity_id == name) + row = conn.execute(stmt).first() + if row: + ds_proto = DataSourceProto.FromString(row["data_source_proto"]) + return DataSource.from_proto(ds_proto) + raise DataSourceObjectNotFoundException(name, project=project) + + def list_data_sources( + self, project: str, allow_cache: bool = False + ) -> List[DataSource]: + with self.engine.connect() as conn: + stmt = select(data_sources) + rows = conn.execute(stmt).all() + if rows: + return [ + DataSource.from_proto( + DataSourceProto.FromString(row["data_source_proto"]) + ) + for row in rows + ] + return [] + + def apply_data_source( + self, data_source: DataSource, project: str, commit: bool = True + ): + with self.engine.connect() as conn: + stmt = select(data_sources).where(entities.c.entity_id == data_source.name) + row = conn.execute(stmt).first() + update_time = int(datetime.utcnow().timestamp()) + if row: + update_stmt = ( + update(data_sources) + .where(data_sources.c.entity_id == data_source.name,) + .values( + entity_proto=data_source.to_proto().SerializeToString(), + last_updated_timestamp=update_time, + ) + ) + conn.execute(update_stmt) + else: + insert_stmt = insert(data_sources).values( + entity_id=data_source.name, + entity_proto=data_source.to_proto().SerializeToString(), + last_updated_timestamp=update_time, + ) + conn.execute(insert_stmt) + + def delete_data_source(self, name: str, project: str, commit: bool = True): + with self.engine.connect() as conn: + stmt = delete(data_sources).where(data_sources.c.entity_id == name) + rows = conn.execute(stmt) + if rows.rowcount < 1: + raise DataSourceObjectNotFoundException(name, project) diff --git a/sdk/python/feast/repo_config.py b/sdk/python/feast/repo_config.py index b7fd9c20377..578ca250dea 100644 --- a/sdk/python/feast/repo_config.py +++ b/sdk/python/feast/repo_config.py @@ -82,6 +82,10 @@ class Config: class RegistryConfig(FeastBaseModel): """Metadata Store Configuration. Configuration that relates to reading from and writing to the Feast registry.""" + registry_type: Optional[StrictStr] + """ str: Provider name or a class name that implements RegistryStore. + If specified, registry_store_type should be redundant.""" + registry_store_type: Optional[StrictStr] """ str: Provider name or a class name that implements RegistryStore. """ diff --git a/setup.py b/setup.py index 633267dbae5..16409de1207 100644 --- a/setup.py +++ b/setup.py @@ -66,6 +66,7 @@ "pydantic>=1,<2", "pygments==2.12.0", "PyYAML>=5.4.*,<7", + "SQLAlchemy[mypy]>1,<2", "tabulate==0.8.*", "tenacity>=7,<9", "toml==0.10.*", From f9e6bb0028a9ea6a10777a855b751ee1c428c35f Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Fri, 27 May 2022 12:57:12 -0700 Subject: [PATCH 02/17] hack hack hack Signed-off-by: Achal Shah --- sdk/python/feast/feature_store.py | 8 +- sdk/python/feast/infra/registry_stores/sql.py | 216 +++++++++++++++--- sdk/python/feast/repo_config.py | 2 +- 3 files changed, 197 insertions(+), 29 deletions(-) diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index f959504826f..ff72c654797 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -71,6 +71,7 @@ ) from feast.infra.infra_object import Infra from feast.infra.provider import Provider, RetrievalJob, get_provider +from feast.infra.registry_stores.sql import SqlRegistry from feast.on_demand_feature_view import OnDemandFeatureView from feast.online_response import OnlineResponse from feast.protos.feast.core.InfraObject_pb2 import Infra as InfraProto @@ -138,8 +139,11 @@ def __init__( raise ValueError("Please specify one of repo_path or config.") registry_config = self.config.get_registry_config() - self._registry = Registry(registry_config, repo_path=self.repo_path) - self._registry._initialize_registry() + if registry_config.reqistry_type == "sql": + self._registry = SqlRegistry(registry_config, None) + else: + self._registry = Registry(registry_config, repo_path=self.repo_path) + self._registry._initialize_registry() self._provider = get_provider(self.config, self.repo_path) self._go_server = None diff --git a/sdk/python/feast/infra/registry_stores/sql.py b/sdk/python/feast/infra/registry_stores/sql.py index 0feee8e5ab4..4683ff7a56b 100644 --- a/sdk/python/feast/infra/registry_stores/sql.py +++ b/sdk/python/feast/infra/registry_stores/sql.py @@ -1,11 +1,12 @@ from datetime import datetime from pathlib import Path +from threading import Lock from typing import List, Optional -from sql import ( # type: ignore - BIGINT, - VARBINARY, +from sqlalchemy import ( # type: ignore + BigInteger, Column, + LargeBinary, MetaData, String, Table, @@ -15,15 +16,31 @@ select, update, ) -from sql.engine import Engine +from sqlalchemy.engine import Engine +from feast.base_feature_view import BaseFeatureView from feast.data_source import DataSource from feast.entity import Entity from feast.errors import DataSourceObjectNotFoundException, EntityNotFoundException +from feast.feature_service import FeatureService +from feast.feature_view import FeatureView +from feast.on_demand_feature_view import OnDemandFeatureView from feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto from feast.protos.feast.core.Entity_pb2 import Entity as EntityProto +from feast.protos.feast.core.FeatureService_pb2 import ( + FeatureService as FeatureServiceProto, +) +from feast.protos.feast.core.FeatureView_pb2 import FeatureView as FeatureViewProto +from feast.protos.feast.core.OnDemandFeatureView_pb2 import ( + OnDemandFeatureView as OnDemandFeatureViewProto, +) +from feast.protos.feast.core.RequestFeatureView_pb2 import ( + RequestFeatureView as RequestFeatureViewProto, +) from feast.registry import Registry from feast.repo_config import RegistryConfig +from feast.request_feature_view import RequestFeatureView +from feast.saved_dataset import SavedDataset metadata = MetaData() @@ -31,57 +48,57 @@ "entities", metadata, Column("entity_id", String, primary_key=True), - Column("last_updated_timestamp", BIGINT, nullable=False), - Column("entity_proto", VARBINARY, nullable=False), + Column("last_updated_timestamp", BigInteger, nullable=False), + Column("entity_proto", LargeBinary, nullable=False), ) data_sources = Table( "data_sources", metadata, Column("data_source_name", String, primary_key=True), - Column("last_updated_timestamp", BIGINT, nullable=False), - Column("data_source_proto", VARBINARY, nullable=False), + Column("last_updated_timestamp", BigInteger, nullable=False), + Column("data_source_proto", LargeBinary, nullable=False), ) feature_views = Table( "feature_views", metadata, Column("feature_view_name", String, primary_key=True), - Column("last_updated_timestamp", BIGINT, nullable=False), - Column("materialized_intervals", VARBINARY, nullable=False), - Column("feature_view_proto", VARBINARY, nullable=False), + Column("last_updated_timestamp", BigInteger, nullable=False), + Column("materialized_intervals", LargeBinary, nullable=True), + Column("feature_view_proto", LargeBinary, nullable=False), ) request_feature_views = Table( "request_feature_views", metadata, Column("feature_view_name", String, primary_key=True), - Column("last_updated_timestamp", BIGINT, nullable=False), - Column("feature_view_proto", VARBINARY, nullable=False), + Column("last_updated_timestamp", BigInteger, nullable=False), + Column("feature_view_proto", LargeBinary, nullable=False), ) on_demand_feature_views = Table( "on_demand_feature_views", metadata, Column("feature_view_name", String, primary_key=True), - Column("last_updated_timestamp", BIGINT, nullable=False), - Column("feature_view_proto", VARBINARY, nullable=False), + Column("last_updated_timestamp", BigInteger, nullable=False), + Column("feature_view_proto", LargeBinary, nullable=False), ) feature_user_metadata = Table( "feature_metadata", metadata, Column("feature_name", String, primary_key=True), - Column("last_updated_timestamp", BIGINT, nullable=False), - Column("feature_metadata_binary", VARBINARY, nullable=False), + Column("last_updated_timestamp", BigInteger, nullable=False), + Column("feature_metadata_binary", LargeBinary, nullable=False), ) feature_services = Table( "feature_services", metadata, Column("feature_service_name", String, primary_key=True), - Column("last_updated_timestamp", BIGINT, nullable=False), - Column("feature_service_proto", VARBINARY, nullable=False), + Column("last_updated_timestamp", BigInteger, nullable=False), + Column("feature_service_proto", LargeBinary, nullable=False), ) APPLY_OPERATIONS = {"entity": (entities, entities.c.entity_id)} @@ -92,11 +109,24 @@ def __init__( self, registry_config: Optional[RegistryConfig], repo_path: Optional[Path] ): assert registry_config - self.engine: Engine = create_engine(registry_config.path, echo=True) + self.engine: Engine = create_engine(registry_config.path, echo=False) metadata.create_all(self.engine) + self._refresh_lock = Lock() def teardown(self): - super().teardown() + for t in { + feature_views, + feature_services, + data_sources, + on_demand_feature_views, + request_feature_views, + }: + with self.engine.connect() as conn: + stmt = delete(t) + conn.execute(stmt) + + def refresh(self): + pass def apply_entity(self, entity: Entity, project: str, commit: bool = True): with self.engine.connect() as conn: @@ -182,23 +212,92 @@ def apply_data_source( self, data_source: DataSource, project: str, commit: bool = True ): with self.engine.connect() as conn: - stmt = select(data_sources).where(entities.c.entity_id == data_source.name) + stmt = select(data_sources).where( + data_sources.c.data_source_name == data_source.name + ) row = conn.execute(stmt).first() update_time = int(datetime.utcnow().timestamp()) if row: update_stmt = ( update(data_sources) - .where(data_sources.c.entity_id == data_source.name,) + .where(data_sources.c.data_source_name == data_source.name,) .values( - entity_proto=data_source.to_proto().SerializeToString(), + data_source_proto=data_source.to_proto().SerializeToString(), last_updated_timestamp=update_time, ) ) conn.execute(update_stmt) else: insert_stmt = insert(data_sources).values( - entity_id=data_source.name, - entity_proto=data_source.to_proto().SerializeToString(), + data_source_name=data_source.name, + data_source_proto=data_source.to_proto().SerializeToString(), + last_updated_timestamp=update_time, + ) + conn.execute(insert_stmt) + + def apply_feature_view( + self, feature_view: BaseFeatureView, project: str, commit: bool = True + ): + if isinstance(feature_view, FeatureView): + fv_table = feature_views + elif isinstance(feature_view, OnDemandFeatureView): + fv_table = on_demand_feature_views + elif isinstance(feature_view, RequestFeatureView): + fv_table = request_feature_views + else: + raise ValueError(f"Unexpected feature view type: {type(feature_view)}") + + with self.engine.connect() as conn: + stmt = select(fv_table).where( + fv_table.c.feature_view_name == feature_view.name + ) + row = conn.execute(stmt).first() + feature_view.last_updated_timestamp = datetime.utcnow() + update_time = int(feature_view.last_updated_timestamp.timestamp()) + if row: + update_stmt = ( + update(fv_table) + .where(fv_table.c.feature_view_name == feature_view.name,) + .values( + feature_view_proto=feature_view.to_proto().SerializeToString(), + last_updated_timestamp=update_time, + ) + ) + conn.execute(update_stmt) + else: + insert_stmt = insert(fv_table).values( + feature_view_name=feature_view.name, + feature_view_proto=feature_view.to_proto().SerializeToString(), + last_updated_timestamp=update_time, + ) + conn.execute(insert_stmt) + + def apply_feature_service( + self, feature_service: FeatureService, project: str, commit: bool = True + ): + with self.engine.connect() as conn: + stmt = select(feature_services).where( + feature_services.c.feature_service_name == feature_service.name + ) + row = conn.execute(stmt).first() + feature_service.last_updated_timestamp = datetime.utcnow() + update_time = int(feature_service.last_updated_timestamp.timestamp()) + if row: + update_stmt = ( + update(feature_services) + .where( + feature_services.c.feature_service_name == feature_service.name, + ) + .values( + feature_service_proto=feature_service.to_proto().SerializeToString(), + last_updated_timestamp=update_time, + ) + ) + conn.execute(update_stmt) + else: + insert_stmt = insert(feature_services).values( + feature_service_name=feature_service.name, + feature_service_proto=feature_service.to_proto().SerializeToString(), last_updated_timestamp=update_time, ) conn.execute(insert_stmt) @@ -209,3 +308,68 @@ def delete_data_source(self, name: str, project: str, commit: bool = True): rows = conn.execute(stmt) if rows.rowcount < 1: raise DataSourceObjectNotFoundException(name, project) + + def list_feature_services( + self, project: str, allow_cache: bool = False + ) -> List[FeatureService]: + with self.engine.connect() as conn: + stmt = select(feature_services) + rows = conn.execute(stmt).all() + if rows: + return [ + FeatureService.from_proto( + FeatureServiceProto.FromString(row["feature_service_proto"]) + ) + for row in rows + ] + return [] + + def list_feature_views( + self, project: str, allow_cache: bool = False + ) -> List[FeatureView]: + with self.engine.connect() as conn: + stmt = select(feature_views) + rows = conn.execute(stmt).all() + if rows: + return [ + FeatureView.from_proto( + FeatureViewProto.FromString(row["feature_view_proto"]) + ) + for row in rows + ] + return [] + + def list_saved_datasets( + self, project: str, allow_cache: bool = False + ) -> List[SavedDataset]: + return [] + + def list_request_feature_views( + self, project: str, allow_cache: bool = False + ) -> List[RequestFeatureView]: + with self.engine.connect() as conn: + stmt = select(request_feature_views) + rows = conn.execute(stmt).all() + if rows: + return [ + RequestFeatureView.from_proto( + RequestFeatureViewProto.FromString(row["feature_view_proto"]) + ) + for row in rows + ] + return [] + + def list_on_demand_feature_views( + self, project: str, allow_cache: bool = False + ) -> List[OnDemandFeatureView]: + with self.engine.connect() as conn: + stmt = select(on_demand_feature_views) + rows = conn.execute(stmt).all() + if rows: + return [ + OnDemandFeatureView.from_proto( + OnDemandFeatureViewProto.FromString(row["feature_view_proto"]) + ) + for row in rows + ] + return [] diff --git a/sdk/python/feast/repo_config.py b/sdk/python/feast/repo_config.py index 578ca250dea..4583feed350 100644 --- a/sdk/python/feast/repo_config.py +++ b/sdk/python/feast/repo_config.py @@ -82,7 +82,7 @@ class Config: class RegistryConfig(FeastBaseModel): """Metadata Store Configuration. Configuration that relates to reading from and writing to the Feast registry.""" - registry_type: Optional[StrictStr] + reqistry_type: StrictStr = "file" """ str: Provider name or a class name that implements RegistryStore. If specified, registry_store_type should be redundant.""" From 99a6b3ecd4aca5a062a21e23468b00c0b2d5d7a1 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Fri, 27 May 2022 16:59:03 -0700 Subject: [PATCH 03/17] fix repo config Signed-off-by: Achal Shah --- sdk/python/feast/feature_store.py | 2 +- sdk/python/feast/repo_config.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index ff72c654797..c1c228b9646 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -139,7 +139,7 @@ def __init__( raise ValueError("Please specify one of repo_path or config.") registry_config = self.config.get_registry_config() - if registry_config.reqistry_type == "sql": + if registry_config.registry_type == "sql": self._registry = SqlRegistry(registry_config, None) else: self._registry = Registry(registry_config, repo_path=self.repo_path) diff --git a/sdk/python/feast/repo_config.py b/sdk/python/feast/repo_config.py index 4583feed350..b7cf1683dc6 100644 --- a/sdk/python/feast/repo_config.py +++ b/sdk/python/feast/repo_config.py @@ -82,7 +82,7 @@ class Config: class RegistryConfig(FeastBaseModel): """Metadata Store Configuration. Configuration that relates to reading from and writing to the Feast registry.""" - reqistry_type: StrictStr = "file" + registry_type: StrictStr = "file" """ str: Provider name or a class name that implements RegistryStore. If specified, registry_store_type should be redundant.""" From 64c397123b93819faa7fb64177afa4aa3e75b5d4 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Tue, 31 May 2022 14:25:44 -0700 Subject: [PATCH 04/17] reduce duplication Signed-off-by: Achal Shah --- sdk/python/feast/infra/registry_stores/sql.py | 201 +++++++++--------- 1 file changed, 100 insertions(+), 101 deletions(-) diff --git a/sdk/python/feast/infra/registry_stores/sql.py b/sdk/python/feast/infra/registry_stores/sql.py index 4683ff7a56b..99ffea2dd33 100644 --- a/sdk/python/feast/infra/registry_stores/sql.py +++ b/sdk/python/feast/infra/registry_stores/sql.py @@ -37,10 +37,11 @@ from feast.protos.feast.core.RequestFeatureView_pb2 import ( RequestFeatureView as RequestFeatureViewProto, ) +from feast.protos.feast.core.SavedDataset_pb2 import SavedDataset as SavedDatasetProto from feast.registry import Registry from feast.repo_config import RegistryConfig from feast.request_feature_view import RequestFeatureView -from feast.saved_dataset import SavedDataset +from feast.saved_dataset import SavedDataset, ValidationReference metadata = MetaData() @@ -101,16 +102,34 @@ Column("feature_service_proto", LargeBinary, nullable=False), ) -APPLY_OPERATIONS = {"entity": (entities, entities.c.entity_id)} +saved_datasets = Table( + "saved_datasets", + metadata, + Column("saved_dataset_name", String, primary_key=True), + Column("last_updated_timestamp", BigInteger, nullable=False), + Column("saved_dataset_proto", LargeBinary, nullable=False), +) + +validation_references = Table( + "validation_references", + metadata, + Column("validation_reference_name", String, primary_key=True), + Column("last_updated_timestamp", BigInteger, nullable=False), + Column("validation_reference_proto", LargeBinary, nullable=False), +) class SqlRegistry(Registry): def __init__( self, registry_config: Optional[RegistryConfig], repo_path: Optional[Path] ): - assert registry_config + assert registry_config is not None, "SqlRegistry needs a valid registry_config" self.engine: Engine = create_engine(registry_config.path, echo=False) metadata.create_all(self.engine) + + # _refresh_lock is not used by the SqlRegistry, but is present to conform to the + # Registry class. + # TODO: remove external references to _refresh_lock and remove field. self._refresh_lock = Lock() def teardown(self): @@ -120,6 +139,8 @@ def teardown(self): data_sources, on_demand_feature_views, request_feature_views, + saved_datasets, + validation_references, }: with self.engine.connect() as conn: stmt = delete(t) @@ -129,31 +150,7 @@ def refresh(self): pass def apply_entity(self, entity: Entity, project: str, commit: bool = True): - with self.engine.connect() as conn: - stmt = select(entities).where(entities.c.entity_id == entity.name) - entity.last_updated_timestamp = datetime.utcnow() - row = conn.execute(stmt).first() - if row: - update_stmt = ( - update(entities) - .where(entities.c.entity_id == entity.name,) - .values( - entity_proto=entity.to_proto().SerializeToString(), - last_updated_timestamp=int( - entity.last_updated_timestamp.timestamp() - ), - ) - ) - conn.execute(update_stmt) - else: - insert_stmt = insert(entities).values( - entity_id=entity.name, - entity_proto=entity.to_proto().SerializeToString(), - last_updated_timestamp=int( - entity.last_updated_timestamp.timestamp() - ), - ) - conn.execute(insert_stmt) + return self._apply_object(entities, "entity_id", entity, "entity_proto") def get_entity(self, name: str, project: str, allow_cache: bool = False) -> Entity: with self.engine.connect() as conn: @@ -211,29 +208,9 @@ def list_data_sources( def apply_data_source( self, data_source: DataSource, project: str, commit: bool = True ): - with self.engine.connect() as conn: - stmt = select(data_sources).where( - data_sources.c.data_source_name == data_source.name - ) - row = conn.execute(stmt).first() - update_time = int(datetime.utcnow().timestamp()) - if row: - update_stmt = ( - update(data_sources) - .where(data_sources.c.data_source_name == data_source.name,) - .values( - data_source_proto=data_source.to_proto().SerializeToString(), - last_updated_timestamp=update_time, - ) - ) - conn.execute(update_stmt) - else: - insert_stmt = insert(data_sources).values( - data_source_name=data_source.name, - data_source_proto=data_source.to_proto().SerializeToString(), - last_updated_timestamp=update_time, - ) - conn.execute(insert_stmt) + return self._apply_object( + data_sources, "data_source_name", data_source, "data_source_proto" + ) def apply_feature_view( self, feature_view: BaseFeatureView, project: str, commit: bool = True @@ -247,60 +224,19 @@ def apply_feature_view( else: raise ValueError(f"Unexpected feature view type: {type(feature_view)}") - with self.engine.connect() as conn: - stmt = select(fv_table).where( - fv_table.c.feature_view_name == feature_view.name - ) - row = conn.execute(stmt).first() - feature_view.last_updated_timestamp = datetime.utcnow() - update_time = int(feature_view.last_updated_timestamp.timestamp()) - if row: - update_stmt = ( - update(fv_table) - .where(fv_table.c.feature_view_name == feature_view.name,) - .values( - feature_view_proto=feature_view.to_proto().SerializeToString(), - last_updated_timestamp=update_time, - ) - ) - conn.execute(update_stmt) - else: - insert_stmt = insert(fv_table).values( - feature_view_name=feature_view.name, - feature_view_proto=feature_view.to_proto().SerializeToString(), - last_updated_timestamp=update_time, - ) - conn.execute(insert_stmt) + return self._apply_object( + fv_table, "feature_view_name", feature_view, "feature_view_proto" + ) def apply_feature_service( self, feature_service: FeatureService, project: str, commit: bool = True ): - with self.engine.connect() as conn: - stmt = select(feature_services).where( - feature_services.c.feature_service_name == feature_service.name - ) - row = conn.execute(stmt).first() - feature_service.last_updated_timestamp = datetime.utcnow() - update_time = int(feature_service.last_updated_timestamp.timestamp()) - if row: - update_stmt = ( - update(feature_services) - .where( - feature_services.c.feature_service_name == feature_service.name, - ) - .values( - feature_service_proto=feature_service.to_proto().SerializeToString(), - last_updated_timestamp=update_time, - ) - ) - conn.execute(update_stmt) - else: - insert_stmt = insert(feature_services).values( - feature_service_name=feature_service.name, - feature_service_proto=feature_service.to_proto().SerializeToString(), - last_updated_timestamp=update_time, - ) - conn.execute(insert_stmt) + return self._apply_object( + feature_services, + "feature_service_name", + feature_service, + "feature_service_proto", + ) def delete_data_source(self, name: str, project: str, commit: bool = True): with self.engine.connect() as conn: @@ -342,6 +278,16 @@ def list_feature_views( def list_saved_datasets( self, project: str, allow_cache: bool = False ) -> List[SavedDataset]: + with self.engine.connect() as conn: + stmt = select(saved_datasets) + rows = conn.execute(stmt).all() + if rows: + return [ + SavedDataset.from_proto( + SavedDatasetProto.FromString(row["saved_dataset_proto"]) + ) + for row in rows + ] return [] def list_request_feature_views( @@ -373,3 +319,56 @@ def list_on_demand_feature_views( for row in rows ] return [] + + def apply_saved_dataset( + self, saved_dataset: SavedDataset, project: str, commit: bool = True, + ): + return self._apply_object( + saved_datasets, "saved_dataset_name", saved_dataset, "saved_dataset_proto" + ) + + def apply_validation_reference( + self, + validation_reference: ValidationReference, + project: str, + commit: bool = True, + ): + return self._apply_object( + validation_references, + "validation_reference_name", + validation_reference, + "validation_reference_proto", + ) + + def _apply_object( + self, table, id_field_name, obj, proto_field_name, + ): + name = obj.name + with self.engine.connect() as conn: + stmt = select(table).where(getattr(table.c, id_field_name) == name) + row = conn.execute(stmt).first() + update_datetime = datetime.utcnow() + update_time = int(update_datetime.timestamp()) + if hasattr(obj, "last_updated_timestamp"): + obj.last_updated_timestamp = update_datetime + if row: + update_stmt = ( + update(table) + .where(getattr(table.c, id_field_name) == name) + .values( + **{ + proto_field_name: obj.to_proto().SerializeToString(), + "last_updated_timestamp": update_time, + }, + ) + ) + conn.execute(update_stmt) + else: + insert_stmt = insert(feature_services).values( + **{ + id_field_name: name, + proto_field_name: obj.to_proto().SerializeToString(), + "last_updated_timestamp": update_time, + }, + ) + conn.execute(insert_stmt) From dcd88bdbd2b6f700bc928fb646e93f1190671bf7 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 2 Jun 2022 10:21:15 -0700 Subject: [PATCH 05/17] simplify Signed-off-by: Achal Shah --- sdk/python/feast/infra/registry_stores/sql.py | 116 +++++++----------- 1 file changed, 41 insertions(+), 75 deletions(-) diff --git a/sdk/python/feast/infra/registry_stores/sql.py b/sdk/python/feast/infra/registry_stores/sql.py index 99ffea2dd33..7c0f0f1d2c0 100644 --- a/sdk/python/feast/infra/registry_stores/sql.py +++ b/sdk/python/feast/infra/registry_stores/sql.py @@ -162,15 +162,7 @@ def get_entity(self, name: str, project: str, allow_cache: bool = False) -> Enti raise EntityNotFoundException(name, project=project) def list_entities(self, project: str, allow_cache: bool = False) -> List[Entity]: - with self.engine.connect() as conn: - stmt = select(entities) - rows = conn.execute(stmt).all() - if rows: - return [ - Entity.from_proto(EntityProto.FromString(row["entity_proto"])) - for row in rows - ] - return [] + return self._list_objects(entities, EntityProto, Entity, "entity_proto") def delete_entity(self, name: str, project: str, commit: bool = True): with self.engine.connect() as conn: @@ -193,17 +185,9 @@ def get_data_source( def list_data_sources( self, project: str, allow_cache: bool = False ) -> List[DataSource]: - with self.engine.connect() as conn: - stmt = select(data_sources) - rows = conn.execute(stmt).all() - if rows: - return [ - DataSource.from_proto( - DataSourceProto.FromString(row["data_source_proto"]) - ) - for row in rows - ] - return [] + return self._list_objects( + data_sources, DataSourceProto, DataSource, "data_source_proto" + ) def apply_data_source( self, data_source: DataSource, project: str, commit: bool = True @@ -248,77 +232,46 @@ def delete_data_source(self, name: str, project: str, commit: bool = True): def list_feature_services( self, project: str, allow_cache: bool = False ) -> List[FeatureService]: - with self.engine.connect() as conn: - stmt = select(feature_services) - rows = conn.execute(stmt).all() - if rows: - return [ - FeatureService.from_proto( - FeatureServiceProto.FromString(row["feature_service_proto"]) - ) - for row in rows - ] - return [] + return self._list_objects( + feature_services, + FeatureServiceProto, + FeatureService, + "feature_service_proto", + ) def list_feature_views( self, project: str, allow_cache: bool = False ) -> List[FeatureView]: - with self.engine.connect() as conn: - stmt = select(feature_views) - rows = conn.execute(stmt).all() - if rows: - return [ - FeatureView.from_proto( - FeatureViewProto.FromString(row["feature_view_proto"]) - ) - for row in rows - ] - return [] + return self._list_objects( + feature_views, FeatureViewProto, FeatureView, "feature_view_proto" + ) def list_saved_datasets( self, project: str, allow_cache: bool = False ) -> List[SavedDataset]: - with self.engine.connect() as conn: - stmt = select(saved_datasets) - rows = conn.execute(stmt).all() - if rows: - return [ - SavedDataset.from_proto( - SavedDatasetProto.FromString(row["saved_dataset_proto"]) - ) - for row in rows - ] - return [] + return self._list_objects( + saved_datasets, SavedDatasetProto, SavedDataset, "saved_dataset_proto" + ) def list_request_feature_views( self, project: str, allow_cache: bool = False ) -> List[RequestFeatureView]: - with self.engine.connect() as conn: - stmt = select(request_feature_views) - rows = conn.execute(stmt).all() - if rows: - return [ - RequestFeatureView.from_proto( - RequestFeatureViewProto.FromString(row["feature_view_proto"]) - ) - for row in rows - ] - return [] + return self._list_objects( + request_feature_views, + RequestFeatureViewProto, + RequestFeatureView, + "feature_view_proto", + ) def list_on_demand_feature_views( self, project: str, allow_cache: bool = False ) -> List[OnDemandFeatureView]: - with self.engine.connect() as conn: - stmt = select(on_demand_feature_views) - rows = conn.execute(stmt).all() - if rows: - return [ - OnDemandFeatureView.from_proto( - OnDemandFeatureViewProto.FromString(row["feature_view_proto"]) - ) - for row in rows - ] - return [] + return self._list_objects( + on_demand_feature_views, + OnDemandFeatureViewProto, + OnDemandFeatureView, + "feature_view_proto", + ) def apply_saved_dataset( self, saved_dataset: SavedDataset, project: str, commit: bool = True, @@ -372,3 +325,16 @@ def _apply_object( }, ) conn.execute(insert_stmt) + + def _list_objects(self, table, proto_class, python_class, proto_field_name): + with self.engine.connect() as conn: + stmt = select(table) + rows = conn.execute(stmt).all() + if rows: + return [ + python_class.from_proto( + proto_class.FromString(row[proto_field_name]) + ) + for row in rows + ] + return [] From 9add91d55664f93679ff1904eeb0d814bf629f11 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 2 Jun 2022 12:36:07 -0700 Subject: [PATCH 06/17] postgres tests Signed-off-by: Achal Shah --- sdk/python/feast/infra/registry_stores/sql.py | 101 ++++++- .../registration/test_sql_registry.py | 270 ++++++++++++++++++ 2 files changed, 357 insertions(+), 14 deletions(-) create mode 100644 sdk/python/tests/integration/registration/test_sql_registry.py diff --git a/sdk/python/feast/infra/registry_stores/sql.py b/sdk/python/feast/infra/registry_stores/sql.py index 7c0f0f1d2c0..3ba3ed01960 100644 --- a/sdk/python/feast/infra/registry_stores/sql.py +++ b/sdk/python/feast/infra/registry_stores/sql.py @@ -21,7 +21,12 @@ from feast.base_feature_view import BaseFeatureView from feast.data_source import DataSource from feast.entity import Entity -from feast.errors import DataSourceObjectNotFoundException, EntityNotFoundException +from feast.errors import ( + DataSourceObjectNotFoundException, + EntityNotFoundException, + FeatureServiceNotFoundException, + FeatureViewNotFoundException, +) from feast.feature_service import FeatureService from feast.feature_view import FeatureView from feast.on_demand_feature_view import OnDemandFeatureView @@ -161,6 +166,57 @@ def get_entity(self, name: str, project: str, allow_cache: bool = False) -> Enti return Entity.from_proto(entity_proto) raise EntityNotFoundException(name, project=project) + def get_feature_view( + self, name: str, project: str, allow_cache: bool = False + ) -> FeatureView: + with self.engine.connect() as conn: + stmt = select(feature_views).where( + feature_views.c.feature_view_name == name + ) + row = conn.execute(stmt).first() + if row: + fv_proto = FeatureViewProto.FromString(row["feature_view_proto"]) + return FeatureView.from_proto(fv_proto) + raise FeatureViewNotFoundException(name, project=project) + + def get_on_demand_feature_view( + self, name: str, project: str, allow_cache: bool = False + ) -> OnDemandFeatureView: + with self.engine.connect() as conn: + stmt = select(on_demand_feature_views).where( + on_demand_feature_views.c.feature_view_name == name + ) + row = conn.execute(stmt).first() + if row: + fv_proto = OnDemandFeatureViewProto.FromString( + row["feature_view_proto"] + ) + return OnDemandFeatureView.from_proto(fv_proto) + raise FeatureViewNotFoundException(name, project=project) + + def get_feature_service( + self, name: str, project: str, allow_cache: bool = False + ) -> FeatureService: + with self.engine.connect() as conn: + stmt = select(feature_services).where( + feature_services.c.feature_service_name == name + ) + row = conn.execute(stmt).first() + if row: + fv_proto = FeatureServiceProto.FromString(row["feature_service_proto"]) + return FeatureService.from_proto(fv_proto) + raise FeatureServiceNotFoundException(name, project=project) + + def get_saved_dataset( + self, name: str, project: str, allow_cache: bool = False + ) -> SavedDataset: + pass + + def get_validation_reference( + self, name: str, project: str, allow_cache: bool = False + ) -> ValidationReference: + pass + def list_entities(self, project: str, allow_cache: bool = False) -> List[Entity]: return self._list_objects(entities, EntityProto, Entity, "entity_proto") @@ -171,6 +227,25 @@ def delete_entity(self, name: str, project: str, commit: bool = True): if rows.rowcount < 1: raise EntityNotFoundException(name, project) + def delete_feature_view(self, name: str, project: str, commit: bool = True): + deleted_count = 0 + for table in {feature_views, request_feature_views, on_demand_feature_views}: + with self.engine.connect() as conn: + stmt = delete(table).where(table.c.feature_view_name == name) + rows = conn.execute(stmt) + deleted_count += rows.rowcount + if deleted_count == 0: + raise FeatureViewNotFoundException(name, project) + + def delete_feature_service(self, name: str, project: str, commit: bool = True): + with self.engine.connect() as conn: + stmt = delete(feature_services).where( + feature_services.c.feature_service_name == name + ) + rows = conn.execute(stmt) + if rows.rowcount < 1: + raise FeatureServiceNotFoundException(name, project) + def get_data_source( self, name: str, project: str, allow_cache: bool = False ) -> DataSource: @@ -305,25 +380,23 @@ def _apply_object( if hasattr(obj, "last_updated_timestamp"): obj.last_updated_timestamp = update_datetime if row: + values = { + proto_field_name: obj.to_proto().SerializeToString(), + "last_updated_timestamp": update_time, + } update_stmt = ( update(table) .where(getattr(table.c, id_field_name) == name) - .values( - **{ - proto_field_name: obj.to_proto().SerializeToString(), - "last_updated_timestamp": update_time, - }, - ) + .values(values,) ) conn.execute(update_stmt) else: - insert_stmt = insert(feature_services).values( - **{ - id_field_name: name, - proto_field_name: obj.to_proto().SerializeToString(), - "last_updated_timestamp": update_time, - }, - ) + values = { + id_field_name: name, + proto_field_name: obj.to_proto().SerializeToString(), + "last_updated_timestamp": update_time, + } + insert_stmt = insert(table).values(values,) conn.execute(insert_stmt) def _list_objects(self, table, proto_class, python_class, proto_field_name): diff --git a/sdk/python/tests/integration/registration/test_sql_registry.py b/sdk/python/tests/integration/registration/test_sql_registry.py new file mode 100644 index 00000000000..34b01f81492 --- /dev/null +++ b/sdk/python/tests/integration/registration/test_sql_registry.py @@ -0,0 +1,270 @@ +# Copyright 2021 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. +import logging +from datetime import timedelta + +import pandas as pd +import pytest +from testcontainers.core.container import DockerContainer +from testcontainers.core.waiting_utils import wait_for_logs + +from feast import FileSource +from feast.data_format import ParquetFormat +from feast.entity import Entity +from feast.feature_view import FeatureView +from feast.field import Field +from feast.infra.registry_stores.sql import SqlRegistry +from feast.on_demand_feature_view import on_demand_feature_view +from feast.repo_config import RegistryConfig +from feast.types import Array, Bytes, Float32, Int64, String + +POSTGRES_USER = "test" +POSTGRES_PASSWORD = "test" +POSTGRES_DB = "test" + + +logger = logging.getLogger(__name__) + + +@pytest.fixture(scope="session") +def sql_registry(): + container = ( + DockerContainer("postgres:latest") + .with_exposed_ports(5432) + .with_env("POSTGRES_USER", POSTGRES_USER) + .with_env("POSTGRES_PASSWORD", POSTGRES_PASSWORD) + .with_env("POSTGRES_DB", POSTGRES_DB) + ) + + container.start() + + log_string_to_wait_for = "database system is ready to accept connections" + waited = wait_for_logs( + container=container, predicate=log_string_to_wait_for, timeout=30, interval=10, + ) + logger.info("Waited for %s seconds until postgres container was up", waited) + container_port = container.get_exposed_port(5432) + + registry_config = RegistryConfig( + registry_type="sql", + path=f"postgresql://{POSTGRES_USER}:{POSTGRES_PASSWORD}@127.0.0.1:{container_port}/{POSTGRES_DB}", + ) + + yield SqlRegistry(registry_config, None) + + container.stop() + + +def test_apply_entity_success(sql_registry): + entity = Entity( + name="driver_car_id", description="Car driver id", tags={"team": "matchmaking"}, + ) + + project = "project" + + # Register Entity + sql_registry.apply_entity(entity, project) + + entities = sql_registry.list_entities(project) + + entity = entities[0] + assert ( + len(entities) == 1 + and entity.name == "driver_car_id" + and entity.description == "Car driver id" + and "team" in entity.tags + and entity.tags["team"] == "matchmaking" + ) + + entity = sql_registry.get_entity("driver_car_id", project) + assert ( + entity.name == "driver_car_id" + and entity.description == "Car driver id" + and "team" in entity.tags + and entity.tags["team"] == "matchmaking" + ) + + sql_registry.delete_entity("driver_car_id", project) + entities = sql_registry.list_entities(project) + assert len(entities) == 0 + + sql_registry.teardown() + + +@pytest.mark.integration +def test_apply_entity_integration(sql_registry): + entity = Entity( + name="driver_car_id", description="Car driver id", tags={"team": "matchmaking"}, + ) + + project = "project" + + # Register Entity + sql_registry.apply_entity(entity, project) + + entities = sql_registry.list_entities(project) + + entity = entities[0] + assert ( + len(entities) == 1 + and entity.name == "driver_car_id" + and entity.description == "Car driver id" + and "team" in entity.tags + and entity.tags["team"] == "matchmaking" + ) + + entity = sql_registry.get_entity("driver_car_id", project) + assert ( + entity.name == "driver_car_id" + and entity.description == "Car driver id" + and "team" in entity.tags + and entity.tags["team"] == "matchmaking" + ) + + sql_registry.teardown() + + +def test_apply_feature_view_success(sql_registry): + # Create Feature Views + batch_source = FileSource( + file_format=ParquetFormat(), + path="file://feast/*", + timestamp_field="ts_col", + created_timestamp_column="timestamp", + ) + + entity = Entity(name="fs1_my_entity_1", join_keys=["test"]) + + fv1 = FeatureView( + name="my_feature_view_1", + schema=[ + Field(name="fs1_my_feature_1", dtype=Int64), + Field(name="fs1_my_feature_2", dtype=String), + Field(name="fs1_my_feature_3", dtype=Array(String)), + Field(name="fs1_my_feature_4", dtype=Array(Bytes)), + ], + entities=[entity], + tags={"team": "matchmaking"}, + batch_source=batch_source, + ttl=timedelta(minutes=5), + ) + + project = "project" + + # Register Feature View + sql_registry.apply_feature_view(fv1, project) + + feature_views = sql_registry.list_feature_views(project) + + # List Feature Views + assert ( + len(feature_views) == 1 + and feature_views[0].name == "my_feature_view_1" + and feature_views[0].features[0].name == "fs1_my_feature_1" + and feature_views[0].features[0].dtype == Int64 + and feature_views[0].features[1].name == "fs1_my_feature_2" + and feature_views[0].features[1].dtype == String + and feature_views[0].features[2].name == "fs1_my_feature_3" + and feature_views[0].features[2].dtype == Array(String) + and feature_views[0].features[3].name == "fs1_my_feature_4" + and feature_views[0].features[3].dtype == Array(Bytes) + and feature_views[0].entities[0] == "fs1_my_entity_1" + ) + + feature_view = sql_registry.get_feature_view("my_feature_view_1", project) + assert ( + feature_view.name == "my_feature_view_1" + and feature_view.features[0].name == "fs1_my_feature_1" + and feature_view.features[0].dtype == Int64 + and feature_view.features[1].name == "fs1_my_feature_2" + and feature_view.features[1].dtype == String + and feature_view.features[2].name == "fs1_my_feature_3" + and feature_view.features[2].dtype == Array(String) + and feature_view.features[3].name == "fs1_my_feature_4" + and feature_view.features[3].dtype == Array(Bytes) + and feature_view.entities[0] == "fs1_my_entity_1" + ) + + sql_registry.delete_feature_view("my_feature_view_1", project) + feature_views = sql_registry.list_feature_views(project) + assert len(feature_views) == 0 + + sql_registry.teardown() + + +def test_apply_on_demand_feature_view_success(sql_registry): + # Create Feature Views + driver_stats = FileSource( + name="driver_stats_source", + path="data/driver_stats_lat_lon.parquet", + timestamp_field="event_timestamp", + created_timestamp_column="created", + description="A table describing the stats of a driver based on hourly logs", + owner="test2@gmail.com", + ) + + driver_daily_features_view = FeatureView( + name="driver_daily_features", + entities=["driver"], + ttl=timedelta(seconds=8640000000), + schema=[ + Field(name="daily_miles_driven", dtype=Float32), + Field(name="lat", dtype=Float32), + Field(name="lon", dtype=Float32), + Field(name="string_feature", dtype=String), + ], + online=True, + source=driver_stats, + tags={"production": "True"}, + owner="test2@gmail.com", + ) + + @on_demand_feature_view( + sources=[driver_daily_features_view], + schema=[Field(name="first_char", dtype=String)], + ) + def location_features_from_push(inputs: pd.DataFrame) -> pd.DataFrame: + df = pd.DataFrame() + df["first_char"] = inputs["string_feature"].str[:1].astype("string") + return df + + project = "project" + + # Register Feature View + sql_registry.apply_feature_view(location_features_from_push, project) + + feature_views = sql_registry.list_on_demand_feature_views(project) + + # List Feature Views + assert ( + len(feature_views) == 1 + and feature_views[0].name == "location_features_from_push" + and feature_views[0].features[0].name == "first_char" + and feature_views[0].features[0].dtype == String + ) + + feature_view = sql_registry.get_on_demand_feature_view( + "location_features_from_push", project + ) + assert ( + feature_view.name == "location_features_from_push" + and feature_view.features[0].name == "first_char" + and feature_view.features[0].dtype == String + ) + + sql_registry.delete_feature_view("location_features_from_push", project) + feature_views = sql_registry.list_on_demand_feature_views(project) + assert len(feature_views) == 0 + + sql_registry.teardown() From b7e33eeb6a7eec084b78ce0ff9159e743ac2751e Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 2 Jun 2022 13:33:16 -0700 Subject: [PATCH 07/17] tests for pg and mysql Signed-off-by: Achal Shah --- sdk/python/feast/infra/registry_stores/sql.py | 18 +- .../registration/test_sql_registry.py | 303 +++++++++++++++++- setup.py | 1 + 3 files changed, 310 insertions(+), 12 deletions(-) diff --git a/sdk/python/feast/infra/registry_stores/sql.py b/sdk/python/feast/infra/registry_stores/sql.py index 3ba3ed01960..9d19c44439a 100644 --- a/sdk/python/feast/infra/registry_stores/sql.py +++ b/sdk/python/feast/infra/registry_stores/sql.py @@ -53,7 +53,7 @@ entities = Table( "entities", metadata, - Column("entity_id", String, primary_key=True), + Column("entity_id", String(50), primary_key=True), Column("last_updated_timestamp", BigInteger, nullable=False), Column("entity_proto", LargeBinary, nullable=False), ) @@ -61,7 +61,7 @@ data_sources = Table( "data_sources", metadata, - Column("data_source_name", String, primary_key=True), + Column("data_source_name", String(50), primary_key=True), Column("last_updated_timestamp", BigInteger, nullable=False), Column("data_source_proto", LargeBinary, nullable=False), ) @@ -69,7 +69,7 @@ feature_views = Table( "feature_views", metadata, - Column("feature_view_name", String, primary_key=True), + Column("feature_view_name", String(50), primary_key=True), Column("last_updated_timestamp", BigInteger, nullable=False), Column("materialized_intervals", LargeBinary, nullable=True), Column("feature_view_proto", LargeBinary, nullable=False), @@ -78,7 +78,7 @@ request_feature_views = Table( "request_feature_views", metadata, - Column("feature_view_name", String, primary_key=True), + Column("feature_view_name", String(50), primary_key=True), Column("last_updated_timestamp", BigInteger, nullable=False), Column("feature_view_proto", LargeBinary, nullable=False), ) @@ -86,7 +86,7 @@ on_demand_feature_views = Table( "on_demand_feature_views", metadata, - Column("feature_view_name", String, primary_key=True), + Column("feature_view_name", String(50), primary_key=True), Column("last_updated_timestamp", BigInteger, nullable=False), Column("feature_view_proto", LargeBinary, nullable=False), ) @@ -94,7 +94,7 @@ feature_user_metadata = Table( "feature_metadata", metadata, - Column("feature_name", String, primary_key=True), + Column("feature_name", String(50), primary_key=True), Column("last_updated_timestamp", BigInteger, nullable=False), Column("feature_metadata_binary", LargeBinary, nullable=False), ) @@ -102,7 +102,7 @@ feature_services = Table( "feature_services", metadata, - Column("feature_service_name", String, primary_key=True), + Column("feature_service_name", String(50), primary_key=True), Column("last_updated_timestamp", BigInteger, nullable=False), Column("feature_service_proto", LargeBinary, nullable=False), ) @@ -110,7 +110,7 @@ saved_datasets = Table( "saved_datasets", metadata, - Column("saved_dataset_name", String, primary_key=True), + Column("saved_dataset_name", String(50), primary_key=True), Column("last_updated_timestamp", BigInteger, nullable=False), Column("saved_dataset_proto", LargeBinary, nullable=False), ) @@ -118,7 +118,7 @@ validation_references = Table( "validation_references", metadata, - Column("validation_reference_name", String, primary_key=True), + Column("validation_reference_name", String(50), primary_key=True), Column("last_updated_timestamp", BigInteger, nullable=False), Column("validation_reference_proto", LargeBinary, nullable=False), ) diff --git a/sdk/python/tests/integration/registration/test_sql_registry.py b/sdk/python/tests/integration/registration/test_sql_registry.py index 34b01f81492..f966b457585 100644 --- a/sdk/python/tests/integration/registration/test_sql_registry.py +++ b/sdk/python/tests/integration/registration/test_sql_registry.py @@ -16,10 +16,11 @@ import pandas as pd import pytest +from pytest_lazyfixture import lazy_fixture from testcontainers.core.container import DockerContainer from testcontainers.core.waiting_utils import wait_for_logs -from feast import FileSource +from feast import Feature, FileSource, RequestSource from feast.data_format import ParquetFormat from feast.entity import Entity from feast.feature_view import FeatureView @@ -27,7 +28,8 @@ from feast.infra.registry_stores.sql import SqlRegistry from feast.on_demand_feature_view import on_demand_feature_view from feast.repo_config import RegistryConfig -from feast.types import Array, Bytes, Float32, Int64, String +from feast.types import Array, Bytes, Float32, Int32, Int64, String +from feast.value_type import ValueType POSTGRES_USER = "test" POSTGRES_PASSWORD = "test" @@ -38,7 +40,7 @@ @pytest.fixture(scope="session") -def sql_registry(): +def pg_registry(): container = ( DockerContainer("postgres:latest") .with_exposed_ports(5432) @@ -66,6 +68,39 @@ def sql_registry(): container.stop() +@pytest.fixture(scope="session") +def mysql_registry(): + container = ( + DockerContainer("mysql:latest") + .with_exposed_ports(3306) + .with_env("MYSQL_RANDOM_ROOT_PASSWORD", "true") + .with_env("MYSQL_USER", POSTGRES_USER) + .with_env("MYSQL_PASSWORD", POSTGRES_PASSWORD) + .with_env("MYSQL_DATABASE", POSTGRES_DB) + ) + + container.start() + + log_string_to_wait_for = "/usr/sbin/mysqld: ready for connections. Version: '8.0.29' socket: '/var/run/mysqld/mysqld.sock' port: 3306" + waited = wait_for_logs( + container=container, predicate=log_string_to_wait_for, timeout=30, interval=10, + ) + logger.info("Waited for %s seconds until mysql container was up", waited) + container_port = container.get_exposed_port(3306) + + registry_config = RegistryConfig( + registry_type="sql", + path=f"mysql+mysqldb://{POSTGRES_USER}:{POSTGRES_PASSWORD}@127.0.0.1:{container_port}/{POSTGRES_DB}", + ) + + yield SqlRegistry(registry_config, None) + + container.stop() + + +@pytest.mark.parametrize( + "sql_registry", [lazy_fixture("mysql_registry"), lazy_fixture("pg_registry")], +) def test_apply_entity_success(sql_registry): entity = Entity( name="driver_car_id", description="Car driver id", tags={"team": "matchmaking"}, @@ -103,6 +138,9 @@ def test_apply_entity_success(sql_registry): @pytest.mark.integration +@pytest.mark.parametrize( + "sql_registry", [lazy_fixture("mysql_registry"), lazy_fixture("pg_registry")], +) def test_apply_entity_integration(sql_registry): entity = Entity( name="driver_car_id", description="Car driver id", tags={"team": "matchmaking"}, @@ -135,6 +173,9 @@ def test_apply_entity_integration(sql_registry): sql_registry.teardown() +@pytest.mark.parametrize( + "sql_registry", [lazy_fixture("mysql_registry"), lazy_fixture("pg_registry")], +) def test_apply_feature_view_success(sql_registry): # Create Feature Views batch_source = FileSource( @@ -203,6 +244,9 @@ def test_apply_feature_view_success(sql_registry): sql_registry.teardown() +@pytest.mark.parametrize( + "sql_registry", [lazy_fixture("mysql_registry"), lazy_fixture("pg_registry")], +) def test_apply_on_demand_feature_view_success(sql_registry): # Create Feature Views driver_stats = FileSource( @@ -268,3 +312,256 @@ def location_features_from_push(inputs: pd.DataFrame) -> pd.DataFrame: assert len(feature_views) == 0 sql_registry.teardown() + + +# TODO(kevjumba): remove this in feast 0.23 when deprecating +@pytest.mark.parametrize( + "sql_registry", [lazy_fixture("mysql_registry"), lazy_fixture("pg_registry")], +) +@pytest.mark.parametrize( + "request_source_schema", + [[Field(name="my_input_1", dtype=Int32)], {"my_input_1": ValueType.INT32}], +) +def test_modify_feature_views_success(sql_registry, request_source_schema): + # Create Feature Views + batch_source = FileSource( + file_format=ParquetFormat(), + path="file://feast/*", + timestamp_field="ts_col", + created_timestamp_column="timestamp", + ) + + request_source = RequestSource(name="request_source", schema=request_source_schema,) + + entity = Entity(name="fs1_my_entity_1", join_keys=["test"]) + + fv1 = FeatureView( + name="my_feature_view_1", + schema=[Field(name="fs1_my_feature_1", dtype=Int64)], + entities=[entity], + tags={"team": "matchmaking"}, + batch_source=batch_source, + ttl=timedelta(minutes=5), + ) + + @on_demand_feature_view( + features=[ + Feature(name="odfv1_my_feature_1", dtype=ValueType.STRING), + Feature(name="odfv1_my_feature_2", dtype=ValueType.INT32), + ], + sources=[request_source], + ) + def odfv1(feature_df: pd.DataFrame) -> pd.DataFrame: + data = pd.DataFrame() + data["odfv1_my_feature_1"] = feature_df["my_input_1"].astype("category") + data["odfv1_my_feature_2"] = feature_df["my_input_1"].astype("int32") + return data + + project = "project" + + # Register Feature Views + sql_registry.apply_feature_view(odfv1, project) + sql_registry.apply_feature_view(fv1, project) + + # Modify odfv by changing a single feature dtype + @on_demand_feature_view( + features=[ + Feature(name="odfv1_my_feature_1", dtype=ValueType.FLOAT), + Feature(name="odfv1_my_feature_2", dtype=ValueType.INT32), + ], + sources=[request_source], + ) + def odfv1(feature_df: pd.DataFrame) -> pd.DataFrame: + data = pd.DataFrame() + data["odfv1_my_feature_1"] = feature_df["my_input_1"].astype("float") + data["odfv1_my_feature_2"] = feature_df["my_input_1"].astype("int32") + return data + + # Apply the modified odfv + sql_registry.apply_feature_view(odfv1, project) + + # Check odfv + on_demand_feature_views = sql_registry.list_on_demand_feature_views(project) + + assert ( + len(on_demand_feature_views) == 1 + and on_demand_feature_views[0].name == "odfv1" + and on_demand_feature_views[0].features[0].name == "odfv1_my_feature_1" + and on_demand_feature_views[0].features[0].dtype == Float32 + and on_demand_feature_views[0].features[1].name == "odfv1_my_feature_2" + and on_demand_feature_views[0].features[1].dtype == Int32 + ) + request_schema = on_demand_feature_views[0].get_request_data_schema() + assert ( + list(request_schema.keys())[0] == "my_input_1" + and list(request_schema.values())[0] == ValueType.INT32 + ) + + feature_view = sql_registry.get_on_demand_feature_view("odfv1", project) + assert ( + feature_view.name == "odfv1" + and feature_view.features[0].name == "odfv1_my_feature_1" + and feature_view.features[0].dtype == Float32 + and feature_view.features[1].name == "odfv1_my_feature_2" + and feature_view.features[1].dtype == Int32 + ) + request_schema = feature_view.get_request_data_schema() + assert ( + list(request_schema.keys())[0] == "my_input_1" + and list(request_schema.values())[0] == ValueType.INT32 + ) + + # Make sure fv1 is untouched + feature_views = sql_registry.list_feature_views(project) + + # List Feature Views + assert ( + len(feature_views) == 1 + and feature_views[0].name == "my_feature_view_1" + and feature_views[0].features[0].name == "fs1_my_feature_1" + and feature_views[0].features[0].dtype == Int64 + and feature_views[0].entities[0] == "fs1_my_entity_1" + ) + + feature_view = sql_registry.get_feature_view("my_feature_view_1", project) + assert ( + feature_view.name == "my_feature_view_1" + and feature_view.features[0].name == "fs1_my_feature_1" + and feature_view.features[0].dtype == Int64 + and feature_view.entities[0] == "fs1_my_entity_1" + ) + + sql_registry.teardown() + + +@pytest.mark.integration +@pytest.mark.parametrize( + "sql_registry", [lazy_fixture("mysql_registry"), lazy_fixture("pg_registry")], +) +def test_apply_feature_view_integration(sql_registry): + # Create Feature Views + batch_source = FileSource( + file_format=ParquetFormat(), + path="file://feast/*", + timestamp_field="ts_col", + created_timestamp_column="timestamp", + ) + + entity = Entity(name="fs1_my_entity_1", join_keys=["test"]) + + fv1 = FeatureView( + name="my_feature_view_1", + schema=[ + Field(name="fs1_my_feature_1", dtype=Int64), + Field(name="fs1_my_feature_2", dtype=String), + Field(name="fs1_my_feature_3", dtype=Array(String)), + Field(name="fs1_my_feature_4", dtype=Array(Bytes)), + ], + entities=[entity], + tags={"team": "matchmaking"}, + batch_source=batch_source, + ttl=timedelta(minutes=5), + ) + + project = "project" + + # Register Feature View + sql_registry.apply_feature_view(fv1, project) + + feature_views = sql_registry.list_feature_views(project) + + # List Feature Views + assert ( + len(feature_views) == 1 + and feature_views[0].name == "my_feature_view_1" + and feature_views[0].features[0].name == "fs1_my_feature_1" + and feature_views[0].features[0].dtype == Int64 + and feature_views[0].features[1].name == "fs1_my_feature_2" + and feature_views[0].features[1].dtype == String + and feature_views[0].features[2].name == "fs1_my_feature_3" + and feature_views[0].features[2].dtype == Array(String) + and feature_views[0].features[3].name == "fs1_my_feature_4" + and feature_views[0].features[3].dtype == Array(Bytes) + and feature_views[0].entities[0] == "fs1_my_entity_1" + ) + + feature_view = sql_registry.get_feature_view("my_feature_view_1", project) + assert ( + feature_view.name == "my_feature_view_1" + and feature_view.features[0].name == "fs1_my_feature_1" + and feature_view.features[0].dtype == Int64 + and feature_view.features[1].name == "fs1_my_feature_2" + and feature_view.features[1].dtype == String + and feature_view.features[2].name == "fs1_my_feature_3" + and feature_view.features[2].dtype == Array(String) + and feature_view.features[3].name == "fs1_my_feature_4" + and feature_view.features[3].dtype == Array(Bytes) + and feature_view.entities[0] == "fs1_my_entity_1" + ) + + sql_registry.delete_feature_view("my_feature_view_1", project) + feature_views = sql_registry.list_feature_views(project) + assert len(feature_views) == 0 + + sql_registry.teardown() + + +@pytest.mark.integration +@pytest.mark.parametrize( + "sql_registry", [lazy_fixture("mysql_registry"), lazy_fixture("pg_registry")], +) +def test_apply_data_source(sql_registry): + # Create Feature Views + batch_source = FileSource( + name="test_source", + file_format=ParquetFormat(), + path="file://feast/*", + timestamp_field="ts_col", + created_timestamp_column="timestamp", + ) + + entity = Entity(name="fs1_my_entity_1", join_keys=["test"]) + + fv1 = FeatureView( + name="my_feature_view_1", + schema=[ + Field(name="fs1_my_feature_1", dtype=Int64), + Field(name="fs1_my_feature_2", dtype=String), + Field(name="fs1_my_feature_3", dtype=Array(String)), + Field(name="fs1_my_feature_4", dtype=Array(Bytes)), + ], + entities=[entity], + tags={"team": "matchmaking"}, + batch_source=batch_source, + ttl=timedelta(minutes=5), + ) + + project = "project" + + # Register data source and feature view + sql_registry.apply_data_source(batch_source, project, commit=False) + sql_registry.apply_feature_view(fv1, project, commit=True) + + registry_feature_views = sql_registry.list_feature_views(project) + registry_data_sources = sql_registry.list_data_sources(project) + assert len(registry_feature_views) == 1 + assert len(registry_data_sources) == 1 + registry_feature_view = registry_feature_views[0] + assert registry_feature_view.batch_source == batch_source + registry_data_source = registry_data_sources[0] + assert registry_data_source == batch_source + + # Check that change to batch source propagates + batch_source.timestamp_field = "new_ts_col" + sql_registry.apply_data_source(batch_source, project, commit=False) + sql_registry.apply_feature_view(fv1, project, commit=True) + registry_feature_views = sql_registry.list_feature_views(project) + registry_data_sources = sql_registry.list_data_sources(project) + assert len(registry_feature_views) == 1 + assert len(registry_data_sources) == 1 + registry_feature_view = registry_feature_views[0] + assert registry_feature_view.batch_source == batch_source + registry_batch_source = sql_registry.list_data_sources(project)[0] + assert registry_batch_source == batch_source + + sql_registry.teardown() diff --git a/setup.py b/setup.py index 16409de1207..a9499924eba 100644 --- a/setup.py +++ b/setup.py @@ -132,6 +132,7 @@ "moto", "mypy==0.931", "mypy-protobuf==3.1", + "mysqlclient", "avro==1.10.0", "gcsfs>=0.4.0,<=2022.01.0", "urllib3>=1.25.4,<2", From 4b92e9baa01b79f63167257b16fd7f1bcfbe1919 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 2 Jun 2022 14:21:46 -0700 Subject: [PATCH 08/17] fix requirements Signed-off-by: Achal Shah --- .../requirements/py3.7-ci-requirements.txt | 32 +++++++++++++------ .../requirements/py3.7-requirements.txt | 24 +++++++++++--- .../requirements/py3.8-ci-requirements.txt | 29 +++++++++++------ .../requirements/py3.8-requirements.txt | 19 +++++++++-- .../requirements/py3.9-ci-requirements.txt | 29 +++++++++-------- .../requirements/py3.9-requirements.txt | 21 +++++++++--- setup.py | 9 ++++-- 7 files changed, 118 insertions(+), 45 deletions(-) diff --git a/sdk/python/requirements/py3.7-ci-requirements.txt b/sdk/python/requirements/py3.7-ci-requirements.txt index ce24e767b4e..3be4356c318 100644 --- a/sdk/python/requirements/py3.7-ci-requirements.txt +++ b/sdk/python/requirements/py3.7-ci-requirements.txt @@ -4,7 +4,7 @@ # # pip-compile --extra=ci --output-file=sdk/python/requirements/py3.7-ci-requirements.txt # -absl-py==1.0.0 +absl-py==1.1.0 # via tensorflow-metadata adal==1.2.7 # via @@ -58,7 +58,7 @@ attrs==21.4.0 # pytest avro==1.10.0 # via feast (setup.py) -azure-core==1.24.0 +azure-core==1.24.1 # via # adlfs # azure-identity @@ -126,7 +126,7 @@ colorama==0.4.4 # via # feast (setup.py) # great-expectations -coverage[toml]==6.4 +coverage[toml]==6.4.1 # via pytest-cov cryptography==35.0.0 # via @@ -231,7 +231,7 @@ google-cloud-core==1.7.2 # google-cloud-storage google-cloud-datastore==2.6.1 # via feast (setup.py) -google-cloud-firestore==2.5.1 +google-cloud-firestore==2.5.2 # via firebase-admin google-cloud-storage==1.40.0 # via @@ -251,6 +251,8 @@ googleapis-common-protos==1.56.2 # tensorflow-metadata great-expectations==0.14.13 # via feast (setup.py) +greenlet==1.1.2 + # via sqlalchemy grpcio==1.46.3 # via # feast (setup.py) @@ -300,6 +302,7 @@ importlib-metadata==4.2.0 # pre-commit # pytest # redis + # sqlalchemy # virtualenv importlib-resources==5.7.1 # via jsonschema @@ -328,7 +331,7 @@ jsonpatch==1.32 # via great-expectations jsonpointer==2.3 # via jsonpatch -jsonschema==4.5.1 +jsonschema==4.6.0 # via # altair # feast (setup.py) @@ -375,11 +378,15 @@ multidict==6.0.2 # aiohttp # yarl mypy==0.931 - # via feast (setup.py) + # via + # feast (setup.py) + # sqlalchemy mypy-extensions==0.4.3 # via mypy mypy-protobuf==3.1 # via feast (setup.py) +mysqlclient==2.1.0 + # via feast (setup.py) nbformat==5.4.0 # via great-expectations nodeenv==1.6.0 @@ -454,11 +461,12 @@ proto-plus==1.20.5 # google-cloud-bigquery-storage # google-cloud-datastore # google-cloud-firestore -protobuf==3.19.4 +protobuf==3.20.1 # via # feast (setup.py) # google-api-core # google-cloud-bigquery + # google-cloud-firestore # googleapis-common-protos # grpcio-reflection # grpcio-testing @@ -624,7 +632,6 @@ scipy==1.7.3 # via great-expectations six==1.16.0 # via - # absl-py # azure-core # azure-identity # google-api-core @@ -663,6 +670,10 @@ sphinxcontrib-qthelp==1.0.3 # via sphinx sphinxcontrib-serializinghtml==1.1.5 # via sphinx +sqlalchemy[mypy]==1.4.37 + # via feast (setup.py) +sqlalchemy2-stubs==0.0.2a22 + # via sqlalchemy starlette==0.19.1 # via fastapi tabulate==0.8.9 @@ -718,11 +729,11 @@ types-python-dateutil==2.8.17 # via feast (setup.py) types-pytz==2021.3.8 # via feast (setup.py) -types-pyyaml==6.0.7 +types-pyyaml==6.0.8 # via feast (setup.py) types-redis==4.2.6 # via feast (setup.py) -types-requests==2.27.29 +types-requests==2.27.30 # via feast (setup.py) types-setuptools==57.4.17 # via feast (setup.py) @@ -746,6 +757,7 @@ typing-extensions==4.2.0 # pydantic # redis # responses + # sqlalchemy2-stubs # starlette # uvicorn # yarl diff --git a/sdk/python/requirements/py3.7-requirements.txt b/sdk/python/requirements/py3.7-requirements.txt index 85d3e2ee096..e6658b50a2d 100644 --- a/sdk/python/requirements/py3.7-requirements.txt +++ b/sdk/python/requirements/py3.7-requirements.txt @@ -4,7 +4,7 @@ # # pip-compile --output-file=sdk/python/requirements/py3.7-requirements.txt # -absl-py==1.0.0 +absl-py==1.1.0 # via tensorflow-metadata anyio==3.6.1 # via @@ -49,6 +49,8 @@ googleapis-common-protos==1.56.2 # feast (setup.py) # google-api-core # tensorflow-metadata +greenlet==1.1.2 + # via sqlalchemy grpcio==1.46.3 # via # feast (setup.py) @@ -67,11 +69,12 @@ importlib-metadata==4.11.4 # via # click # jsonschema + # sqlalchemy importlib-resources==5.7.1 # via jsonschema jinja2==3.1.2 # via feast (setup.py) -jsonschema==4.5.1 +jsonschema==4.6.0 # via feast (setup.py) locket==1.0.0 # via partd @@ -79,6 +82,10 @@ markupsafe==2.1.1 # via jinja2 mmh3==3.0.0 # via feast (setup.py) +mypy==0.960 + # via sqlalchemy +mypy-extensions==0.4.3 + # via mypy numpy==1.21.6 # via # feast (setup.py) @@ -97,7 +104,7 @@ partd==1.2.0 # via dask proto-plus==1.20.5 # via feast (setup.py) -protobuf==3.19.4 +protobuf==3.20.1 # via # feast (setup.py) # google-api-core @@ -140,13 +147,16 @@ rsa==4.8 # via google-auth six==1.16.0 # via - # absl-py # google-auth # grpcio # pandavro # python-dateutil sniffio==1.2.0 # via anyio +sqlalchemy[mypy]==1.4.37 + # via feast (setup.py) +sqlalchemy2-stubs==0.0.2a22 + # via sqlalchemy starlette==0.19.1 # via fastapi tabulate==0.8.9 @@ -157,12 +167,16 @@ tensorflow-metadata==1.8.0 # via feast (setup.py) toml==0.10.2 # via feast (setup.py) +tomli==2.0.1 + # via mypy toolz==0.11.2 # via # dask # partd tqdm==4.64.0 # via feast (setup.py) +typed-ast==1.5.4 + # via mypy typing-extensions==4.2.0 # via # anyio @@ -170,7 +184,9 @@ typing-extensions==4.2.0 # h11 # importlib-metadata # jsonschema + # mypy # pydantic + # sqlalchemy2-stubs # starlette # uvicorn urllib3==1.26.9 diff --git a/sdk/python/requirements/py3.8-ci-requirements.txt b/sdk/python/requirements/py3.8-ci-requirements.txt index 8de792c0734..f4429c14bc9 100644 --- a/sdk/python/requirements/py3.8-ci-requirements.txt +++ b/sdk/python/requirements/py3.8-ci-requirements.txt @@ -4,7 +4,7 @@ # # pip-compile --extra=ci --output-file=sdk/python/requirements/py3.8-ci-requirements.txt # -absl-py==1.0.0 +absl-py==1.1.0 # via tensorflow-metadata adal==1.2.7 # via @@ -58,7 +58,7 @@ attrs==21.4.0 # pytest avro==1.10.0 # via feast (setup.py) -azure-core==1.24.0 +azure-core==1.24.1 # via # adlfs # azure-identity @@ -126,7 +126,7 @@ colorama==0.4.4 # via # feast (setup.py) # great-expectations -coverage[toml]==6.4 +coverage[toml]==6.4.1 # via pytest-cov cryptography==35.0.0 # via @@ -233,7 +233,7 @@ google-cloud-core==1.7.2 # google-cloud-storage google-cloud-datastore==2.6.1 # via feast (setup.py) -google-cloud-firestore==2.5.1 +google-cloud-firestore==2.5.2 # via firebase-admin google-cloud-storage==1.40.0 # via @@ -253,6 +253,8 @@ googleapis-common-protos==1.56.2 # tensorflow-metadata great-expectations==0.14.13 # via feast (setup.py) +greenlet==1.1.2 + # via sqlalchemy grpcio==1.46.3 # via # feast (setup.py) @@ -318,7 +320,7 @@ jsonpatch==1.32 # via great-expectations jsonpointer==2.3 # via jsonpatch -jsonschema==4.5.1 +jsonschema==4.6.0 # via # altair # feast (setup.py) @@ -365,11 +367,15 @@ multidict==6.0.2 # aiohttp # yarl mypy==0.931 - # via feast (setup.py) + # via + # feast (setup.py) + # sqlalchemy mypy-extensions==0.4.3 # via mypy mypy-protobuf==3.1 # via feast (setup.py) +mysqlclient==2.1.0 + # via feast (setup.py) nbformat==5.4.0 # via great-expectations nodeenv==1.6.0 @@ -449,6 +455,7 @@ protobuf==3.19.4 # feast (setup.py) # google-api-core # google-cloud-bigquery + # google-cloud-firestore # googleapis-common-protos # grpcio-reflection # grpcio-testing @@ -616,7 +623,6 @@ scipy==1.8.1 # via great-expectations six==1.16.0 # via - # absl-py # azure-core # azure-identity # google-api-core @@ -655,6 +661,10 @@ sphinxcontrib-qthelp==1.0.3 # via sphinx sphinxcontrib-serializinghtml==1.1.5 # via sphinx +sqlalchemy[mypy]==1.4.37 + # via feast (setup.py) +sqlalchemy2-stubs==0.0.2a22 + # via sqlalchemy stack-data==0.2.0 # via ipython starlette==0.19.1 @@ -710,11 +720,11 @@ types-python-dateutil==2.8.17 # via feast (setup.py) types-pytz==2021.3.8 # via feast (setup.py) -types-pyyaml==6.0.7 +types-pyyaml==6.0.8 # via feast (setup.py) types-redis==4.2.6 # via feast (setup.py) -types-requests==2.27.29 +types-requests==2.27.30 # via feast (setup.py) types-setuptools==57.4.17 # via feast (setup.py) @@ -729,6 +739,7 @@ typing-extensions==4.2.0 # great-expectations # mypy # pydantic + # sqlalchemy2-stubs # starlette tzdata==2022.1 # via pytz-deprecation-shim diff --git a/sdk/python/requirements/py3.8-requirements.txt b/sdk/python/requirements/py3.8-requirements.txt index ae49676bd03..2c616f40f86 100644 --- a/sdk/python/requirements/py3.8-requirements.txt +++ b/sdk/python/requirements/py3.8-requirements.txt @@ -4,7 +4,7 @@ # # pip-compile --output-file=sdk/python/requirements/py3.8-requirements.txt # -absl-py==1.0.0 +absl-py==1.1.0 # via tensorflow-metadata anyio==3.6.1 # via @@ -49,6 +49,8 @@ googleapis-common-protos==1.56.2 # feast (setup.py) # google-api-core # tensorflow-metadata +greenlet==1.1.2 + # via sqlalchemy grpcio==1.46.3 # via # feast (setup.py) @@ -67,7 +69,7 @@ importlib-resources==5.7.1 # via jsonschema jinja2==3.1.2 # via feast (setup.py) -jsonschema==4.5.1 +jsonschema==4.6.0 # via feast (setup.py) locket==1.0.0 # via partd @@ -75,6 +77,10 @@ markupsafe==2.1.1 # via jinja2 mmh3==3.0.0 # via feast (setup.py) +mypy==0.960 + # via sqlalchemy +mypy-extensions==0.4.3 + # via mypy numpy==1.21.6 # via # feast (setup.py) @@ -136,13 +142,16 @@ rsa==4.8 # via google-auth six==1.16.0 # via - # absl-py # google-auth # grpcio # pandavro # python-dateutil sniffio==1.2.0 # via anyio +sqlalchemy[mypy]==1.4.37 + # via feast (setup.py) +sqlalchemy2-stubs==0.0.2a22 + # via sqlalchemy starlette==0.19.1 # via fastapi tabulate==0.8.9 @@ -153,6 +162,8 @@ tensorflow-metadata==1.8.0 # via feast (setup.py) toml==0.10.2 # via feast (setup.py) +tomli==2.0.1 + # via mypy toolz==0.11.2 # via # dask @@ -161,7 +172,9 @@ tqdm==4.64.0 # via feast (setup.py) typing-extensions==4.2.0 # via + # mypy # pydantic + # sqlalchemy2-stubs # starlette urllib3==1.26.9 # via requests diff --git a/sdk/python/requirements/py3.9-ci-requirements.txt b/sdk/python/requirements/py3.9-ci-requirements.txt index e8741a237ba..ecff16b639b 100644 --- a/sdk/python/requirements/py3.9-ci-requirements.txt +++ b/sdk/python/requirements/py3.9-ci-requirements.txt @@ -1,5 +1,5 @@ # -# This file is autogenerated by pip-compile with python 3.8 +# This file is autogenerated by pip-compile with python 3.9 # To update, run: # # pip-compile --extra=ci --output-file=sdk/python/requirements/py3.9-ci-requirements.txt @@ -73,10 +73,6 @@ babel==2.10.1 # via sphinx backcall==0.2.0 # via ipython -backports-zoneinfo==0.2.1 - # via - # pytz-deprecation-shim - # tzlocal black==19.10b0 # via feast (setup.py) boto3==1.20.23 @@ -126,7 +122,7 @@ colorama==0.4.4 # via # feast (setup.py) # great-expectations -coverage[toml]==6.4 +coverage[toml]==6.4.1 # via pytest-cov cryptography==35.0.0 # via @@ -253,6 +249,8 @@ googleapis-common-protos==1.56.2 # tensorflow-metadata great-expectations==0.14.13 # via feast (setup.py) +greenlet==1.1.2 + # via sqlalchemy grpcio==1.46.3 # via # feast (setup.py) @@ -291,8 +289,6 @@ imagesize==1.3.0 # via sphinx importlib-metadata==4.11.4 # via great-expectations -importlib-resources==5.7.1 - # via jsonschema iniconfig==1.1.1 # via pytest ipython==8.4.0 @@ -365,11 +361,15 @@ multidict==6.0.2 # aiohttp # yarl mypy==0.931 - # via feast (setup.py) + # via + # feast (setup.py) + # sqlalchemy mypy-extensions==0.4.3 # via mypy mypy-protobuf==3.1 # via feast (setup.py) +mysqlclient==2.1.0 + # via feast (setup.py) nbformat==5.4.0 # via great-expectations nodeenv==1.6.0 @@ -444,7 +444,7 @@ proto-plus==1.20.5 # google-cloud-bigquery-storage # google-cloud-datastore # google-cloud-firestore -protobuf==3.19.4 +protobuf==3.20.1 # via # feast (setup.py) # google-api-core @@ -655,6 +655,10 @@ sphinxcontrib-qthelp==1.0.3 # via sphinx sphinxcontrib-serializinghtml==1.1.5 # via sphinx +sqlalchemy[mypy]==1.4.37 + # via feast (setup.py) +sqlalchemy2-stubs==0.0.2a22 + # via sqlalchemy stack-data==0.2.0 # via ipython starlette==0.19.1 @@ -729,6 +733,7 @@ typing-extensions==4.2.0 # great-expectations # mypy # pydantic + # sqlalchemy2-stubs # starlette tzdata==2022.1 # via pytz-deprecation-shim @@ -772,9 +777,7 @@ xmltodict==0.13.0 yarl==1.7.2 # via aiohttp zipp==3.8.0 - # via - # importlib-metadata - # importlib-resources + # via importlib-metadata # The following packages are considered to be unsafe in a requirements file: # pip diff --git a/sdk/python/requirements/py3.9-requirements.txt b/sdk/python/requirements/py3.9-requirements.txt index 577ff2838f9..3f18c4c073e 100644 --- a/sdk/python/requirements/py3.9-requirements.txt +++ b/sdk/python/requirements/py3.9-requirements.txt @@ -4,7 +4,7 @@ # # pip-compile --output-file=sdk/python/requirements/py3.9-requirements.txt # -absl-py==1.0.0 +absl-py==1.1.0 # via tensorflow-metadata anyio==3.6.1 # via @@ -49,6 +49,8 @@ googleapis-common-protos==1.56.2 # feast (setup.py) # google-api-core # tensorflow-metadata +greenlet==1.1.2 + # via sqlalchemy grpcio==1.46.3 # via # feast (setup.py) @@ -65,7 +67,7 @@ idna==3.3 # requests jinja2==3.1.2 # via feast (setup.py) -jsonschema==4.5.1 +jsonschema==4.6.0 # via feast (setup.py) locket==1.0.0 # via partd @@ -73,6 +75,10 @@ markupsafe==2.1.1 # via jinja2 mmh3==3.0.0 # via feast (setup.py) +mypy==0.960 + # via sqlalchemy +mypy-extensions==0.4.3 + # via mypy numpy==1.21.6 # via # feast (setup.py) @@ -91,7 +97,7 @@ partd==1.2.0 # via dask proto-plus==1.20.5 # via feast (setup.py) -protobuf==3.19.4 +protobuf==3.20.1 # via # feast (setup.py) # google-api-core @@ -134,13 +140,16 @@ rsa==4.8 # via google-auth six==1.16.0 # via - # absl-py # google-auth # grpcio # pandavro # python-dateutil sniffio==1.2.0 # via anyio +sqlalchemy[mypy]==1.4.37 + # via feast (setup.py) +sqlalchemy2-stubs==0.0.2a22 + # via sqlalchemy starlette==0.19.1 # via fastapi tabulate==0.8.9 @@ -151,6 +160,8 @@ tensorflow-metadata==1.8.0 # via feast (setup.py) toml==0.10.2 # via feast (setup.py) +tomli==2.0.1 + # via mypy toolz==0.11.2 # via # dask @@ -159,7 +170,9 @@ tqdm==4.64.0 # via feast (setup.py) typing-extensions==4.2.0 # via + # mypy # pydantic + # sqlalchemy2-stubs # starlette urllib3==1.26.9 # via requests diff --git a/setup.py b/setup.py index a9499924eba..62427f9bf6e 100644 --- a/setup.py +++ b/setup.py @@ -60,7 +60,7 @@ "numpy<1.22,<2", "pandas>=1,<2", "pandavro==1.5.*", - "protobuf>=3.10,<3.20", + "protobuf>=3.10,<3.25", "proto-plus==1.20.*", "pyarrow>=4,<7", "pydantic>=1,<2", @@ -108,6 +108,10 @@ "psycopg2-binary>=2.8.3,<3", ] +MYSQL_REQUIRED = [ + "mysqlclient", +] + HBASE_REQUIRED = [ "happybase>=1.2.0,<3", ] @@ -132,7 +136,6 @@ "moto", "mypy==0.931", "mypy-protobuf==3.1", - "mysqlclient", "avro==1.10.0", "gcsfs>=0.4.0,<=2022.01.0", "urllib3>=1.25.4,<2", @@ -169,6 +172,7 @@ + SNOWFLAKE_REQUIRED + SPARK_REQUIRED + POSTGRES_REQUIRED + + MYSQL_REQUIRED + TRINO_REQUIRED + GE_REQUIRED + HBASE_REQUIRED @@ -472,6 +476,7 @@ def copy_extensions_to_source(self): "spark": SPARK_REQUIRED, "trino": TRINO_REQUIRED, "postgres": POSTGRES_REQUIRED, + "mysql": MYSQL_REQUIRED, "ge": GE_REQUIRED, "hbase": HBASE_REQUIRED, "go": GO_REQUIRED, From 7aaf4c018525fa208425d992c1d3e0119d7bb8f8 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 2 Jun 2022 15:11:18 -0700 Subject: [PATCH 09/17] pin protobuf again Signed-off-by: Achal Shah --- .../requirements/py3.10-ci-requirements.txt | 15 +++++++++++++-- sdk/python/requirements/py3.10-requirements.txt | 17 ++++++++++++++++- .../requirements/py3.7-ci-requirements.txt | 2 +- sdk/python/requirements/py3.7-requirements.txt | 2 +- .../requirements/py3.9-ci-requirements.txt | 2 +- sdk/python/requirements/py3.9-requirements.txt | 2 +- setup.py | 2 +- 7 files changed, 34 insertions(+), 8 deletions(-) diff --git a/sdk/python/requirements/py3.10-ci-requirements.txt b/sdk/python/requirements/py3.10-ci-requirements.txt index 0098c057972..bef786ecaa9 100644 --- a/sdk/python/requirements/py3.10-ci-requirements.txt +++ b/sdk/python/requirements/py3.10-ci-requirements.txt @@ -122,7 +122,7 @@ colorama==0.4.4 # via # feast (setup.py) # great-expectations -coverage[toml]==6.4 +coverage[toml]==6.4.1 # via pytest-cov cryptography==35.0.0 # via @@ -249,6 +249,8 @@ googleapis-common-protos==1.56.2 # tensorflow-metadata great-expectations==0.14.13 # via feast (setup.py) +greenlet==1.1.2 + # via sqlalchemy grpcio==1.46.3 # via # feast (setup.py) @@ -359,11 +361,15 @@ multidict==6.0.2 # aiohttp # yarl mypy==0.931 - # via feast (setup.py) + # via + # feast (setup.py) + # sqlalchemy mypy-extensions==0.4.3 # via mypy mypy-protobuf==3.1 # via feast (setup.py) +mysqlclient==2.1.0 + # via feast (setup.py) nbformat==5.4.0 # via great-expectations nodeenv==1.6.0 @@ -647,6 +653,10 @@ sphinxcontrib-qthelp==1.0.3 # via sphinx sphinxcontrib-serializinghtml==1.1.5 # via sphinx +sqlalchemy[mypy]==1.4.37 + # via feast (setup.py) +sqlalchemy2-stubs==0.0.2a22 + # via sqlalchemy stack-data==0.2.0 # via ipython starlette==0.19.1 @@ -720,6 +730,7 @@ typing-extensions==4.2.0 # great-expectations # mypy # pydantic + # sqlalchemy2-stubs tzdata==2022.1 # via pytz-deprecation-shim tzlocal==4.2 diff --git a/sdk/python/requirements/py3.10-requirements.txt b/sdk/python/requirements/py3.10-requirements.txt index 00b14d2cfe7..540455fd99f 100644 --- a/sdk/python/requirements/py3.10-requirements.txt +++ b/sdk/python/requirements/py3.10-requirements.txt @@ -49,6 +49,8 @@ googleapis-common-protos==1.56.2 # feast (setup.py) # google-api-core # tensorflow-metadata +greenlet==1.1.2 + # via sqlalchemy grpcio==1.46.3 # via # feast (setup.py) @@ -73,6 +75,10 @@ markupsafe==2.1.1 # via jinja2 mmh3==3.0.0 # via feast (setup.py) +mypy==0.960 + # via sqlalchemy +mypy-extensions==0.4.3 + # via mypy numpy==1.21.6 # via # feast (setup.py) @@ -140,6 +146,10 @@ six==1.16.0 # python-dateutil sniffio==1.2.0 # via anyio +sqlalchemy[mypy]==1.4.37 + # via feast (setup.py) +sqlalchemy2-stubs==0.0.2a22 + # via sqlalchemy starlette==0.19.1 # via fastapi tabulate==0.8.9 @@ -150,6 +160,8 @@ tensorflow-metadata==1.8.0 # via feast (setup.py) toml==0.10.2 # via feast (setup.py) +tomli==2.0.1 + # via mypy toolz==0.11.2 # via # dask @@ -157,7 +169,10 @@ toolz==0.11.2 tqdm==4.64.0 # via feast (setup.py) typing-extensions==4.2.0 - # via pydantic + # via + # mypy + # pydantic + # sqlalchemy2-stubs urllib3==1.26.9 # via requests uvicorn[standard]==0.17.6 diff --git a/sdk/python/requirements/py3.7-ci-requirements.txt b/sdk/python/requirements/py3.7-ci-requirements.txt index 3be4356c318..1c9e2f37461 100644 --- a/sdk/python/requirements/py3.7-ci-requirements.txt +++ b/sdk/python/requirements/py3.7-ci-requirements.txt @@ -461,7 +461,7 @@ proto-plus==1.20.5 # google-cloud-bigquery-storage # google-cloud-datastore # google-cloud-firestore -protobuf==3.20.1 +protobuf==3.19.4 # via # feast (setup.py) # google-api-core diff --git a/sdk/python/requirements/py3.7-requirements.txt b/sdk/python/requirements/py3.7-requirements.txt index e6658b50a2d..79c5a997977 100644 --- a/sdk/python/requirements/py3.7-requirements.txt +++ b/sdk/python/requirements/py3.7-requirements.txt @@ -104,7 +104,7 @@ partd==1.2.0 # via dask proto-plus==1.20.5 # via feast (setup.py) -protobuf==3.20.1 +protobuf==3.19.4 # via # feast (setup.py) # google-api-core diff --git a/sdk/python/requirements/py3.9-ci-requirements.txt b/sdk/python/requirements/py3.9-ci-requirements.txt index ecff16b639b..156b7656921 100644 --- a/sdk/python/requirements/py3.9-ci-requirements.txt +++ b/sdk/python/requirements/py3.9-ci-requirements.txt @@ -444,7 +444,7 @@ proto-plus==1.20.5 # google-cloud-bigquery-storage # google-cloud-datastore # google-cloud-firestore -protobuf==3.20.1 +protobuf==3.19.4 # via # feast (setup.py) # google-api-core diff --git a/sdk/python/requirements/py3.9-requirements.txt b/sdk/python/requirements/py3.9-requirements.txt index 3f18c4c073e..245542b64f0 100644 --- a/sdk/python/requirements/py3.9-requirements.txt +++ b/sdk/python/requirements/py3.9-requirements.txt @@ -97,7 +97,7 @@ partd==1.2.0 # via dask proto-plus==1.20.5 # via feast (setup.py) -protobuf==3.20.1 +protobuf==3.19.4 # via # feast (setup.py) # google-api-core diff --git a/setup.py b/setup.py index 62427f9bf6e..f92db4acecb 100644 --- a/setup.py +++ b/setup.py @@ -60,7 +60,7 @@ "numpy<1.22,<2", "pandas>=1,<2", "pandavro==1.5.*", - "protobuf>=3.10,<3.25", + "protobuf>=3.10,<3.20", "proto-plus==1.20.*", "pyarrow>=4,<7", "pydantic>=1,<2", From 2555e8a8b8baedac6b55d6394bbb96df5032a279 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 2 Jun 2022 15:27:24 -0700 Subject: [PATCH 10/17] fix macos test Signed-off-by: Achal Shah --- .github/workflows/unit_tests.yml | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/.github/workflows/unit_tests.yml b/.github/workflows/unit_tests.yml index 77198080537..7bea8d1a28a 100644 --- a/.github/workflows/unit_tests.yml +++ b/.github/workflows/unit_tests.yml @@ -8,7 +8,7 @@ jobs: fail-fast: false matrix: python-version: [ "3.7", "3.8", "3.9", "3.10" ] - os: [ ubuntu-latest, macOS-latest] + os: [ ubuntu-latest ] exclude: - os: macOS-latest python-version: "3.8" @@ -32,9 +32,14 @@ jobs: uses: actions/setup-go@v2 with: go-version: 1.17.7 + - name: Install mysql on macOS + if: ${{ matrix.os == "macOS-latest" }} + run: | + brew install mysql + PATH=$PATH:/usr/local/mysql/bin - name: Upgrade pip version run: | - pip install --upgrade "pip>=21.3.1,<22.1" + pip install --upgrade "pip>=22.1,<23" - name: Get pip cache dir id: pip-cache run: | @@ -83,7 +88,7 @@ jobs: python-version: "3.7" - name: Upgrade pip version run: | - pip install --upgrade "pip>=21.3.1,<22.1" + pip install --upgrade "pip>=22.1,<23" - name: Setup Go id: setup-go uses: actions/setup-go@v2 From 29c6b1c680ea3345d5918acf03031b6d42d29db9 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 2 Jun 2022 16:00:10 -0700 Subject: [PATCH 11/17] fix macos test Signed-off-by: Achal Shah --- .github/workflows/unit_tests.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/unit_tests.yml b/.github/workflows/unit_tests.yml index 7bea8d1a28a..c01dff4e3df 100644 --- a/.github/workflows/unit_tests.yml +++ b/.github/workflows/unit_tests.yml @@ -33,7 +33,7 @@ jobs: with: go-version: 1.17.7 - name: Install mysql on macOS - if: ${{ matrix.os == "macOS-latest" }} + if: startsWith(matrix.os, "macOS") run: | brew install mysql PATH=$PATH:/usr/local/mysql/bin From 4b7aed19671f5d001b331809a1b8ea3e99a05e35 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 2 Jun 2022 16:06:23 -0700 Subject: [PATCH 12/17] fix quotes Signed-off-by: Achal Shah --- .github/workflows/unit_tests.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/unit_tests.yml b/.github/workflows/unit_tests.yml index c01dff4e3df..d7fe8da8056 100644 --- a/.github/workflows/unit_tests.yml +++ b/.github/workflows/unit_tests.yml @@ -33,7 +33,7 @@ jobs: with: go-version: 1.17.7 - name: Install mysql on macOS - if: startsWith(matrix.os, "macOS") + if: startsWith(matrix.os, 'macOS') run: | brew install mysql PATH=$PATH:/usr/local/mysql/bin From 16e89442da0ec19d548c9849f3e5989837d195f4 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 2 Jun 2022 16:13:02 -0700 Subject: [PATCH 13/17] add macos Signed-off-by: Achal Shah --- .github/workflows/unit_tests.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/unit_tests.yml b/.github/workflows/unit_tests.yml index d7fe8da8056..ea6141f3310 100644 --- a/.github/workflows/unit_tests.yml +++ b/.github/workflows/unit_tests.yml @@ -8,7 +8,7 @@ jobs: fail-fast: false matrix: python-version: [ "3.7", "3.8", "3.9", "3.10" ] - os: [ ubuntu-latest ] + os: [ ubuntu-latest, macOS-latest ] exclude: - os: macOS-latest python-version: "3.8" From 6a33f4641bfb172040200dcccfabefab56972b43 Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 2 Jun 2022 20:21:09 -0700 Subject: [PATCH 14/17] remove macos Signed-off-by: Achal Shah --- .github/workflows/unit_tests.yml | 5 ---- .../registration/test_sql_registry.py | 23 ++++++++++++++++++- 2 files changed, 22 insertions(+), 6 deletions(-) diff --git a/.github/workflows/unit_tests.yml b/.github/workflows/unit_tests.yml index ea6141f3310..3b0b7f38535 100644 --- a/.github/workflows/unit_tests.yml +++ b/.github/workflows/unit_tests.yml @@ -32,11 +32,6 @@ jobs: uses: actions/setup-go@v2 with: go-version: 1.17.7 - - name: Install mysql on macOS - if: startsWith(matrix.os, 'macOS') - run: | - brew install mysql - PATH=$PATH:/usr/local/mysql/bin - name: Upgrade pip version run: | pip install --upgrade "pip>=22.1,<23" diff --git a/sdk/python/tests/integration/registration/test_sql_registry.py b/sdk/python/tests/integration/registration/test_sql_registry.py index f966b457585..831f967b6df 100644 --- a/sdk/python/tests/integration/registration/test_sql_registry.py +++ b/sdk/python/tests/integration/registration/test_sql_registry.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. import logging +import sys from datetime import timedelta import pandas as pd @@ -98,6 +99,9 @@ def mysql_registry(): container.stop() +@pytest.mark.skipif( + sys.platform == "darwin", reason="does not run on mac github actions" +) @pytest.mark.parametrize( "sql_registry", [lazy_fixture("mysql_registry"), lazy_fixture("pg_registry")], ) @@ -137,6 +141,9 @@ def test_apply_entity_success(sql_registry): sql_registry.teardown() +@pytest.mark.skipif( + sys.platform == "darwin", reason="does not run on mac github actions" +) @pytest.mark.integration @pytest.mark.parametrize( "sql_registry", [lazy_fixture("mysql_registry"), lazy_fixture("pg_registry")], @@ -173,6 +180,9 @@ def test_apply_entity_integration(sql_registry): sql_registry.teardown() +@pytest.mark.skipif( + sys.platform == "darwin", reason="does not run on mac github actions" +) @pytest.mark.parametrize( "sql_registry", [lazy_fixture("mysql_registry"), lazy_fixture("pg_registry")], ) @@ -244,6 +254,9 @@ def test_apply_feature_view_success(sql_registry): sql_registry.teardown() +@pytest.mark.skipif( + sys.platform == "darwin", reason="does not run on mac github actions" +) @pytest.mark.parametrize( "sql_registry", [lazy_fixture("mysql_registry"), lazy_fixture("pg_registry")], ) @@ -314,7 +327,9 @@ def location_features_from_push(inputs: pd.DataFrame) -> pd.DataFrame: sql_registry.teardown() -# TODO(kevjumba): remove this in feast 0.23 when deprecating +@pytest.mark.skipif( + sys.platform == "darwin", reason="does not run on mac github actions" +) @pytest.mark.parametrize( "sql_registry", [lazy_fixture("mysql_registry"), lazy_fixture("pg_registry")], ) @@ -434,6 +449,9 @@ def odfv1(feature_df: pd.DataFrame) -> pd.DataFrame: sql_registry.teardown() +@pytest.mark.skipif( + sys.platform == "darwin", reason="does not run on mac github actions" +) @pytest.mark.integration @pytest.mark.parametrize( "sql_registry", [lazy_fixture("mysql_registry"), lazy_fixture("pg_registry")], @@ -506,6 +524,9 @@ def test_apply_feature_view_integration(sql_registry): sql_registry.teardown() +@pytest.mark.skipif( + sys.platform == "darwin", reason="does not run on mac github actions" +) @pytest.mark.integration @pytest.mark.parametrize( "sql_registry", [lazy_fixture("mysql_registry"), lazy_fixture("pg_registry")], From 629083cdaf29e055a0db5a4e423cf62c3258460f Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Thu, 2 Jun 2022 20:44:09 -0700 Subject: [PATCH 15/17] install mysql library but don't try to run tests Signed-off-by: Achal Shah --- .github/workflows/unit_tests.yml | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/.github/workflows/unit_tests.yml b/.github/workflows/unit_tests.yml index 3b0b7f38535..ea6141f3310 100644 --- a/.github/workflows/unit_tests.yml +++ b/.github/workflows/unit_tests.yml @@ -32,6 +32,11 @@ jobs: uses: actions/setup-go@v2 with: go-version: 1.17.7 + - name: Install mysql on macOS + if: startsWith(matrix.os, 'macOS') + run: | + brew install mysql + PATH=$PATH:/usr/local/mysql/bin - name: Upgrade pip version run: | pip install --upgrade "pip>=22.1,<23" From 4a2c84c8083c43e439cb21c195424040ff9c6c4b Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Fri, 3 Jun 2022 10:15:18 -0700 Subject: [PATCH 16/17] Fix CR comments Signed-off-by: Achal Shah --- sdk/python/feast/infra/registry_stores/sql.py | 143 +++++++++++------- .../registration/test_sql_registry.py | 114 -------------- 2 files changed, 91 insertions(+), 166 deletions(-) diff --git a/sdk/python/feast/infra/registry_stores/sql.py b/sdk/python/feast/infra/registry_stores/sql.py index 9d19c44439a..44ce6d9ff01 100644 --- a/sdk/python/feast/infra/registry_stores/sql.py +++ b/sdk/python/feast/infra/registry_stores/sql.py @@ -26,6 +26,8 @@ EntityNotFoundException, FeatureServiceNotFoundException, FeatureViewNotFoundException, + SavedDatasetNotFound, + ValidationReferenceNotFound, ) from feast.feature_service import FeatureService from feast.feature_view import FeatureView @@ -43,6 +45,9 @@ RequestFeatureView as RequestFeatureViewProto, ) from feast.protos.feast.core.SavedDataset_pb2 import SavedDataset as SavedDatasetProto +from feast.protos.feast.core.ValidationProfile_pb2 import ( + ValidationReference as ValidationReferenceProto, +) from feast.registry import Registry from feast.repo_config import RegistryConfig from feast.request_feature_view import RequestFeatureView @@ -53,7 +58,7 @@ entities = Table( "entities", metadata, - Column("entity_id", String(50), primary_key=True), + Column("entity_name", String(50), primary_key=True), Column("last_updated_timestamp", BigInteger, nullable=False), Column("entity_proto", LargeBinary, nullable=False), ) @@ -91,14 +96,6 @@ Column("feature_view_proto", LargeBinary, nullable=False), ) -feature_user_metadata = Table( - "feature_metadata", - metadata, - Column("feature_name", String(50), primary_key=True), - Column("last_updated_timestamp", BigInteger, nullable=False), - Column("feature_metadata_binary", LargeBinary, nullable=False), -) - feature_services = Table( "feature_services", metadata, @@ -139,9 +136,10 @@ def __init__( def teardown(self): for t in { + entities, + data_sources, feature_views, feature_services, - data_sources, on_demand_feature_views, request_feature_views, saved_datasets, @@ -155,74 +153,96 @@ def refresh(self): pass def apply_entity(self, entity: Entity, project: str, commit: bool = True): - return self._apply_object(entities, "entity_id", entity, "entity_proto") + return self._apply_object(entities, "entity_name", entity, "entity_proto") def get_entity(self, name: str, project: str, allow_cache: bool = False) -> Entity: - with self.engine.connect() as conn: - stmt = select(entities).where(entities.c.entity_id == name) - row = conn.execute(stmt).first() - if row: - entity_proto = EntityProto.FromString(row["entity_proto"]) - return Entity.from_proto(entity_proto) - raise EntityNotFoundException(name, project=project) + return self._get_object( + entities, + name, + project, + EntityProto, + Entity, + "entity_name", + "entity_proto", + EntityNotFoundException, + ) def get_feature_view( self, name: str, project: str, allow_cache: bool = False ) -> FeatureView: - with self.engine.connect() as conn: - stmt = select(feature_views).where( - feature_views.c.feature_view_name == name - ) - row = conn.execute(stmt).first() - if row: - fv_proto = FeatureViewProto.FromString(row["feature_view_proto"]) - return FeatureView.from_proto(fv_proto) - raise FeatureViewNotFoundException(name, project=project) + return self._get_object( + feature_views, + name, + project, + FeatureViewProto, + FeatureView, + "feature_view_name", + "feature_view_proto", + FeatureViewNotFoundException, + ) def get_on_demand_feature_view( self, name: str, project: str, allow_cache: bool = False ) -> OnDemandFeatureView: - with self.engine.connect() as conn: - stmt = select(on_demand_feature_views).where( - on_demand_feature_views.c.feature_view_name == name - ) - row = conn.execute(stmt).first() - if row: - fv_proto = OnDemandFeatureViewProto.FromString( - row["feature_view_proto"] - ) - return OnDemandFeatureView.from_proto(fv_proto) - raise FeatureViewNotFoundException(name, project=project) + return self._get_object( + on_demand_feature_views, + name, + project, + OnDemandFeatureViewProto, + OnDemandFeatureView, + "feature_view_name", + "feature_view_proto", + FeatureViewNotFoundException, + ) def get_feature_service( self, name: str, project: str, allow_cache: bool = False ) -> FeatureService: - with self.engine.connect() as conn: - stmt = select(feature_services).where( - feature_services.c.feature_service_name == name - ) - row = conn.execute(stmt).first() - if row: - fv_proto = FeatureServiceProto.FromString(row["feature_service_proto"]) - return FeatureService.from_proto(fv_proto) - raise FeatureServiceNotFoundException(name, project=project) + return self._get_object( + feature_services, + name, + project, + FeatureServiceProto, + FeatureService, + "feature_service_name", + "feature_service_proto", + FeatureServiceNotFoundException, + ) def get_saved_dataset( self, name: str, project: str, allow_cache: bool = False ) -> SavedDataset: - pass + return self._get_object( + saved_datasets, + name, + project, + SavedDatasetProto, + SavedDataset, + "saved_dataset_name", + "saved_dataset_proto", + SavedDatasetNotFound, + ) def get_validation_reference( self, name: str, project: str, allow_cache: bool = False ) -> ValidationReference: - pass + return self._get_object( + validation_references, + name, + project, + ValidationReferenceProto, + ValidationReference, + "validation_reference_name", + "validation_reference_proto", + ValidationReferenceNotFound, + ) def list_entities(self, project: str, allow_cache: bool = False) -> List[Entity]: return self._list_objects(entities, EntityProto, Entity, "entity_proto") def delete_entity(self, name: str, project: str, commit: bool = True): with self.engine.connect() as conn: - stmt = delete(entities).where(entities.c.entity_id == name) + stmt = delete(entities).where(entities.c.entity_name == name) rows = conn.execute(stmt) if rows.rowcount < 1: raise EntityNotFoundException(name, project) @@ -250,7 +270,7 @@ def get_data_source( self, name: str, project: str, allow_cache: bool = False ) -> DataSource: with self.engine.connect() as conn: - stmt = select(data_sources).where(data_sources.c.entity_id == name) + stmt = select(data_sources).where(data_sources.c.entity_name == name) row = conn.execute(stmt).first() if row: ds_proto = DataSourceProto.FromString(row["data_source_proto"]) @@ -299,7 +319,7 @@ def apply_feature_service( def delete_data_source(self, name: str, project: str, commit: bool = True): with self.engine.connect() as conn: - stmt = delete(data_sources).where(data_sources.c.entity_id == name) + stmt = delete(data_sources).where(data_sources.c.data_source_name == name) rows = conn.execute(stmt) if rows.rowcount < 1: raise DataSourceObjectNotFoundException(name, project) @@ -411,3 +431,22 @@ def _list_objects(self, table, proto_class, python_class, proto_field_name): for row in rows ] return [] + + def _get_object( + self, + table, + name, + project, + proto_class, + python_class, + id_field_name, + proto_field_name, + not_found_exception, + ): + with self.engine.connect() as conn: + stmt = select(table).where(getattr(table.c, id_field_name) == name) + row = conn.execute(stmt).first() + if row: + _proto = proto_class.FromString(row[proto_field_name]) + return python_class.from_proto(_proto) + raise not_found_exception(name, project) diff --git a/sdk/python/tests/integration/registration/test_sql_registry.py b/sdk/python/tests/integration/registration/test_sql_registry.py index 831f967b6df..efad9f2c812 100644 --- a/sdk/python/tests/integration/registration/test_sql_registry.py +++ b/sdk/python/tests/integration/registration/test_sql_registry.py @@ -141,45 +141,6 @@ def test_apply_entity_success(sql_registry): sql_registry.teardown() -@pytest.mark.skipif( - sys.platform == "darwin", reason="does not run on mac github actions" -) -@pytest.mark.integration -@pytest.mark.parametrize( - "sql_registry", [lazy_fixture("mysql_registry"), lazy_fixture("pg_registry")], -) -def test_apply_entity_integration(sql_registry): - entity = Entity( - name="driver_car_id", description="Car driver id", tags={"team": "matchmaking"}, - ) - - project = "project" - - # Register Entity - sql_registry.apply_entity(entity, project) - - entities = sql_registry.list_entities(project) - - entity = entities[0] - assert ( - len(entities) == 1 - and entity.name == "driver_car_id" - and entity.description == "Car driver id" - and "team" in entity.tags - and entity.tags["team"] == "matchmaking" - ) - - entity = sql_registry.get_entity("driver_car_id", project) - assert ( - entity.name == "driver_car_id" - and entity.description == "Car driver id" - and "team" in entity.tags - and entity.tags["team"] == "matchmaking" - ) - - sql_registry.teardown() - - @pytest.mark.skipif( sys.platform == "darwin", reason="does not run on mac github actions" ) @@ -449,81 +410,6 @@ def odfv1(feature_df: pd.DataFrame) -> pd.DataFrame: sql_registry.teardown() -@pytest.mark.skipif( - sys.platform == "darwin", reason="does not run on mac github actions" -) -@pytest.mark.integration -@pytest.mark.parametrize( - "sql_registry", [lazy_fixture("mysql_registry"), lazy_fixture("pg_registry")], -) -def test_apply_feature_view_integration(sql_registry): - # Create Feature Views - batch_source = FileSource( - file_format=ParquetFormat(), - path="file://feast/*", - timestamp_field="ts_col", - created_timestamp_column="timestamp", - ) - - entity = Entity(name="fs1_my_entity_1", join_keys=["test"]) - - fv1 = FeatureView( - name="my_feature_view_1", - schema=[ - Field(name="fs1_my_feature_1", dtype=Int64), - Field(name="fs1_my_feature_2", dtype=String), - Field(name="fs1_my_feature_3", dtype=Array(String)), - Field(name="fs1_my_feature_4", dtype=Array(Bytes)), - ], - entities=[entity], - tags={"team": "matchmaking"}, - batch_source=batch_source, - ttl=timedelta(minutes=5), - ) - - project = "project" - - # Register Feature View - sql_registry.apply_feature_view(fv1, project) - - feature_views = sql_registry.list_feature_views(project) - - # List Feature Views - assert ( - len(feature_views) == 1 - and feature_views[0].name == "my_feature_view_1" - and feature_views[0].features[0].name == "fs1_my_feature_1" - and feature_views[0].features[0].dtype == Int64 - and feature_views[0].features[1].name == "fs1_my_feature_2" - and feature_views[0].features[1].dtype == String - and feature_views[0].features[2].name == "fs1_my_feature_3" - and feature_views[0].features[2].dtype == Array(String) - and feature_views[0].features[3].name == "fs1_my_feature_4" - and feature_views[0].features[3].dtype == Array(Bytes) - and feature_views[0].entities[0] == "fs1_my_entity_1" - ) - - feature_view = sql_registry.get_feature_view("my_feature_view_1", project) - assert ( - feature_view.name == "my_feature_view_1" - and feature_view.features[0].name == "fs1_my_feature_1" - and feature_view.features[0].dtype == Int64 - and feature_view.features[1].name == "fs1_my_feature_2" - and feature_view.features[1].dtype == String - and feature_view.features[2].name == "fs1_my_feature_3" - and feature_view.features[2].dtype == Array(String) - and feature_view.features[3].name == "fs1_my_feature_4" - and feature_view.features[3].dtype == Array(Bytes) - and feature_view.entities[0] == "fs1_my_entity_1" - ) - - sql_registry.delete_feature_view("my_feature_view_1", project) - feature_views = sql_registry.list_feature_views(project) - assert len(feature_views) == 0 - - sql_registry.teardown() - - @pytest.mark.skipif( sys.platform == "darwin", reason="does not run on mac github actions" ) From 9e6d84bfc6a6c34d5cf9e5b56165c88e50ffdfca Mon Sep 17 00:00:00 2001 From: Achal Shah Date: Fri, 3 Jun 2022 10:42:12 -0700 Subject: [PATCH 17/17] add a comment Signed-off-by: Achal Shah --- sdk/python/feast/infra/registry_stores/sql.py | 1 + 1 file changed, 1 insertion(+) diff --git a/sdk/python/feast/infra/registry_stores/sql.py b/sdk/python/feast/infra/registry_stores/sql.py index 44ce6d9ff01..f13428b7377 100644 --- a/sdk/python/feast/infra/registry_stores/sql.py +++ b/sdk/python/feast/infra/registry_stores/sql.py @@ -294,6 +294,7 @@ def apply_data_source( def apply_feature_view( self, feature_view: BaseFeatureView, project: str, commit: bool = True ): + # TODO(achals): Stream feature views need to be supported. if isinstance(feature_view, FeatureView): fv_table = feature_views elif isinstance(feature_view, OnDemandFeatureView):