From 1bf935a4827541bb127bec5b18d1b664b5a52255 Mon Sep 17 00:00:00 2001 From: mbackes Date: Mon, 14 Aug 2023 16:51:21 -0700 Subject: [PATCH 01/12] integration testing for Milvus - tests will fail until all methods are implemented --- milvus/Dockerfile | 12 +++ sdk/python/feast/driver_test_data.py | 2 + .../vectordb/milvus_online_store.py | 10 +-- sdk/python/feast/repo_config.py | 4 - .../milvus_online_store_creator.py | 21 +++-- .../expediagroup/test_milvus_online_store.py | 79 +++++++------------ .../feature_repos/repo_configuration.py | 13 +++ .../feature_repos/universal/feature_views.py | 20 +++++ .../online_store/test_universal_online.py | 41 ++++++++++ 9 files changed, 136 insertions(+), 66 deletions(-) create mode 100644 milvus/Dockerfile diff --git a/milvus/Dockerfile b/milvus/Dockerfile new file mode 100644 index 00000000000..b3e10165607 --- /dev/null +++ b/milvus/Dockerfile @@ -0,0 +1,12 @@ +FROM python:3.11-slim + +RUN python3 -m pip install milvus==2.2.12 + +# this is needed to divert logs to stdout +RUN mkdir -p /root/.milvus.io/milvus-server/2.2.12/logs/ +RUN touch /root/.milvus.io/milvus-server/2.2.12/logs/milvus-stdout.log +RUN touch /root/.milvus.io/milvus-server/2.2.12/logs/milvus-stderr.log +RUN ln -sf /dev/stdout /root/.milvus.io/milvus-server/2.2.12/logs/milvus-stdout.log \ + && ln -sf /dev/stderr /root/.milvus.io/milvus-server/2.2.12/logs/milvus-stderr.log + +CMD ["milvus-server"] \ No newline at end of file diff --git a/sdk/python/feast/driver_test_data.py b/sdk/python/feast/driver_test_data.py index 58c3e8db8fb..296cf07a295 100644 --- a/sdk/python/feast/driver_test_data.py +++ b/sdk/python/feast/driver_test_data.py @@ -187,6 +187,8 @@ def create_customer_daily_profile_df(customers, start_date, end_date) -> pd.Data df_all_customers["lifetime_trip_count"] = np.random.randint( 0, 1000, size=rows ).astype(np.int32) + df_all_customers["profile_embedding"] = [np.random.default_rng().uniform(-100, 200, 50).astype(np.float32) + for _ in range(rows)] # TODO: Remove created timestamp in order to test whether its really optional df_all_customers["created"] = pd.to_datetime(pd.Timestamp.now(tz=None).round("ms")) diff --git a/sdk/python/feast/expediagroup/vectordb/milvus_online_store.py b/sdk/python/feast/expediagroup/vectordb/milvus_online_store.py index 478709af5c3..558a9f64a56 100644 --- a/sdk/python/feast/expediagroup/vectordb/milvus_online_store.py +++ b/sdk/python/feast/expediagroup/vectordb/milvus_online_store.py @@ -63,11 +63,13 @@ def __init__(self, online_config: RepoConfig): def __enter__(self): # Connecting to Milvus logger.info( - f"Connecting to Milvus with alias {self.online_config.alias} and host {self.online_config.host} and default port {self.online_config.port}." + f"Connecting to Milvus with alias {self.online_config.alias} and host {self.online_config.host} and port {self.online_config.port}." ) connections.connect( + alias=self.online_config.alias, host=self.online_config.host, - username=self.online_config.username, + port=self.online_config.port, + user=self.online_config.username, password=self.online_config.password, use_secure=True, ) @@ -158,9 +160,7 @@ def teardown( tables: Sequence[VectorFeatureView], entities: Sequence[Entity], ): - raise NotImplementedError( - "to be implemented in https://jira.expedia.biz/browse/EAPC-7974" - ) + pass def _convert_featureview_schema_to_milvus_readable( self, feast_schema: List[Field], vector_field, vector_field_dimensions diff --git a/sdk/python/feast/repo_config.py b/sdk/python/feast/repo_config.py index 50a537745f3..a6320c55d9b 100644 --- a/sdk/python/feast/repo_config.py +++ b/sdk/python/feast/repo_config.py @@ -203,8 +203,6 @@ def __init__(self, **data: Any): self._offline_config = "redshift" elif data["provider"] == "azure": self._offline_config = "mssql" - elif data["provider"] == "milvus": - self._online_config = "milvus" self._online_store = None if "online_store" in data: @@ -218,8 +216,6 @@ def __init__(self, **data: Any): self._online_config = "dynamodb" elif data["provider"] == "rockset": self._online_config = "rockset" - elif data["provider"] == "milvus": - self._online_config = "milvus" self._batch_engine = None if "batch_engine" in data: diff --git a/sdk/python/tests/expediagroup/milvus_online_store_creator.py b/sdk/python/tests/expediagroup/milvus_online_store_creator.py index ab21761ead6..b71b3750e91 100644 --- a/sdk/python/tests/expediagroup/milvus_online_store_creator.py +++ b/sdk/python/tests/expediagroup/milvus_online_store_creator.py @@ -1,7 +1,11 @@ +import random from typing import Dict from milvus import default_server +from testcontainers.core.container import DockerContainer +from testcontainers.core.waiting_utils import wait_for_logs + from tests.integration.feature_repos.universal.online_store_creator import ( OnlineStoreCreator, ) @@ -10,15 +14,18 @@ class MilvusOnlineStoreCreator(OnlineStoreCreator): def __init__(self, project_name: str, **kwargs): super().__init__(project_name) - self.host = "localhost" - self.port = 19530 - self.server = default_server - self.server.wait_for_started = False + self.container = DockerContainer("milvus/milvus:2.2.12").with_exposed_ports("19530") def create_online_store(self) -> Dict[str, str]: - self.server.start() + self.container.start() + log_string_to_wait_for = "Milvus Proxy successfully initialized and ready to serve!" + wait_for_logs( + container=self.container, predicate=log_string_to_wait_for, timeout=20 + ) + exposed_port = self.container.get_exposed_port("19530") - return {"type": "milvus", "host": self.host, "port": str(self.port)} + return {"alias": "default", "type": "milvus", "host": "localhost", "port": str(exposed_port), + "username": "user", "password": "password"} def teardown(self): - self.server.stop() + self.container.stop() diff --git a/sdk/python/tests/expediagroup/test_milvus_online_store.py b/sdk/python/tests/expediagroup/test_milvus_online_store.py index 1b30ec4da31..a74a7eae986 100644 --- a/sdk/python/tests/expediagroup/test_milvus_online_store.py +++ b/sdk/python/tests/expediagroup/test_milvus_online_store.py @@ -33,7 +33,7 @@ REGION = "us-west-2" HOST = "localhost" PORT = 19530 -ALIAS = "milvus" +ALIAS = "default" SOURCE = FileSource(path="some path") VECTOR_FIELD = "feature1" DIMENSIONS = 10 @@ -54,13 +54,8 @@ def repo_config(): ) -@pytest.fixture -def milvus_online_store(): - return MilvusOnlineStore() - - @pytest.fixture(scope="session") -def milvus_online_setup(): +def embedded_milvus(): # Creating an online store through embedded Milvus for all tests in the class online_store_creator = MilvusOnlineStoreCreator("milvus") online_store_creator.create_online_store() @@ -71,36 +66,26 @@ def milvus_online_setup(): online_store_creator.teardown() -class PymilvusConnectionContext: - def __enter__(self): - # Connecting to Milvus - connections.connect(host=HOST, port=PORT) - - def __exit__(self, exc_type, exc_value, traceback): - # Disconnecting from Milvus - connections.disconnect("milvus") - - class TestMilvusConnectionManager: - def test_connection_manager(self, repo_config, caplog, milvus_online_setup, mocker): + def test_connection_manager(self, repo_config, caplog, mocker): mocker.patch("pymilvus.connections.connect") with MilvusConnectionManager(repo_config.online_store): assert ( - f"Connecting to Milvus with alias {repo_config.online_store.alias} and host {repo_config.online_store.host} and default port {repo_config.online_store.port}." + f"Connecting to Milvus with alias {repo_config.online_store.alias} and host {repo_config.online_store.host} and port {repo_config.online_store.port}." in caplog.text ) connections.connect.assert_called_once_with( + alias=repo_config.online_store.alias, host=repo_config.online_store.host, - username=repo_config.online_store.username, + port=repo_config.online_store.port, + user=repo_config.online_store.username, password=repo_config.online_store.password, use_secure=True, ) - def test_context_manager_exit( - self, repo_config, caplog, milvus_online_setup, mocker - ): + def test_context_manager_exit(self, repo_config, caplog, mocker): # Create a mock for connections.disconnect mock_disconnect = mocker.patch("pymilvus.connections.disconnect") @@ -125,21 +110,24 @@ class TestMilvusOnlineStore: collection_to_write = "Collection2" collection_to_delete = "Collection1" + unavailable_collection = "abc" - def setup_method(self, milvus_online_setup): + @pytest.fixture(autouse=True) + def setup_method(self, repo_config): # Ensuring that the collections created are dropped before the tests are run - with PymilvusConnectionContext(): + with MilvusConnectionManager(repo_config.online_store): # Dropping collections if they exist if utility.has_collection(self.collection_to_delete): utility.drop_collection(self.collection_to_delete) if utility.has_collection(self.collection_to_write): utility.drop_collection(self.collection_to_write) + if utility.has_collection(self.unavailable_collection): + utility.drop_collection(self.unavailable_collection) # Closing the temporary collection to do this - def test_milvus_update_add_collection( - self, repo_config, milvus_online_setup, caplog - ): + yield + def test_milvus_update_add_collection(self, repo_config, caplog, embedded_milvus): feast_schema = [ Field( name="feature2", @@ -205,18 +193,14 @@ def test_milvus_update_add_collection( ) # Here we want to open and check whether the collection was added and then close the connection. - with PymilvusConnectionContext(): + with MilvusConnectionManager(repo_config.online_store): assert utility.has_collection(self.collection_to_write) is True assert ( Collection(self.collection_to_write).schema == schema1 or Collection(self.collection_to_write).schema == schema2 ) - def test_milvus_update_add_existing_collection( - self, repo_config, caplog, milvus_online_setup - ): - - self.setup_method(milvus_online_setup) + def test_milvus_update_add_existing_collection(self, repo_config, caplog, embedded_milvus): # Creating a common schema for collection feast_schema = [ Field( @@ -225,7 +209,7 @@ def test_milvus_update_add_existing_collection( tags={ "is_primary": "False", "description": "float32", - "dimension": "128", + "dimensions": "128", }, ), Field( @@ -248,7 +232,7 @@ def test_milvus_update_add_existing_collection( ) # Here we want to open and add a collection using pymilvus directly and close the connection. - with PymilvusConnectionContext(): + with MilvusConnectionManager(repo_config.online_store): Collection(name=self.collection_to_write, schema=schema) assert utility.has_collection(self.collection_to_write) is True assert len(utility.list_collections()) == 1 @@ -272,14 +256,11 @@ def test_milvus_update_add_existing_collection( ) # Here we want to open and add a collection using pymilvus directly and close the connection, we need to check if the collection count remains 1 and exists. - with PymilvusConnectionContext(): + with MilvusConnectionManager(repo_config.online_store): assert utility.has_collection(self.collection_to_write) is True assert len(utility.list_collections()) == 1 - def test_milvus_update_delete_collection( - self, repo_config, caplog, milvus_online_setup - ): - self.setup_method(milvus_online_setup) + def test_milvus_update_delete_collection(self, repo_config, caplog, embedded_milvus): # Creating a common schema for collection which is compatible with FEAST feast_schema = [ Field( @@ -288,7 +269,7 @@ def test_milvus_update_delete_collection( tags={ "is_primary": "False", "description": "float32", - "dimension": "128", + "dimensions": "128", }, ), Field( @@ -311,7 +292,7 @@ def test_milvus_update_delete_collection( ) # Here we want to open and add a collection using pymilvus directly and close the connection - with PymilvusConnectionContext(): + with MilvusConnectionManager(repo_config.online_store): Collection(name=self.collection_to_write, schema=schema) assert utility.has_collection(self.collection_to_write) is True @@ -334,12 +315,10 @@ def test_milvus_update_delete_collection( ) # Opening and closing the connection and checking if the collection is actually deleted. - with PymilvusConnectionContext(): + with MilvusConnectionManager(repo_config.online_store): assert utility.has_collection(self.collection_to_write) is False - def test_milvus_update_delete_unavailable_collection( - self, repo_config, caplog, milvus_online_setup - ): + def test_milvus_update_delete_unavailable_collection(self, repo_config, caplog, embedded_milvus): feast_schema = [ Field( name="feature1", @@ -347,7 +326,7 @@ def test_milvus_update_delete_unavailable_collection( tags={ "is_primary": "False", "description": "float32", - "dimension": "128", + "dimensions": "128", }, ), Field( @@ -361,7 +340,7 @@ def test_milvus_update_delete_unavailable_collection( config=repo_config, tables_to_delete=[ VectorFeatureView( - name="abc", + name=self.unavailable_collection, schema=feast_schema, source=SOURCE, vector_field=VECTOR_FIELD, @@ -375,5 +354,5 @@ def test_milvus_update_delete_unavailable_collection( partial=None, ) - with PymilvusConnectionContext(): + with MilvusConnectionManager(repo_config.online_store): assert len(utility.list_collections()) == 0 diff --git a/sdk/python/tests/integration/feature_repos/repo_configuration.py b/sdk/python/tests/integration/feature_repos/repo_configuration.py index fda5b3c11de..5c5d08fd895 100644 --- a/sdk/python/tests/integration/feature_repos/repo_configuration.py +++ b/sdk/python/tests/integration/feature_repos/repo_configuration.py @@ -20,6 +20,9 @@ from feast.infra.feature_servers.base_config import FeatureLoggingConfig from feast.infra.feature_servers.local_process.config import LocalFeatureServerConfig from feast.repo_config import RegistryConfig, RepoConfig +from tests.expediagroup.milvus_online_store_creator import ( + MilvusOnlineStoreCreator, +) from tests.integration.feature_repos.integration_test_repo_config import ( IntegrationTestRepoConfig, RegistryLocation, @@ -99,6 +102,15 @@ "host": os.getenv("ROCKSET_APISERVER", "api.rs2.usw2.rockset.com"), } +MILVUS_CONFIG = { + "alias": "default", + "type": "milvus", + "host": "localhost", + "port": 19530, + "username": "user", + "password": "password" +} + OFFLINE_STORE_TO_PROVIDER_CONFIG: Dict[str, DataSourceCreator] = { "file": ("local", FileDataSourceCreator), "bigquery": ("gcp", BigQueryDataSourceCreator), @@ -114,6 +126,7 @@ str, Tuple[Union[str, Dict[str, str]], Optional[Type[OnlineStoreCreator]]] ] = { "sqlite": ({"type": "sqlite"}, None), + "milvus": (MILVUS_CONFIG, MilvusOnlineStoreCreator), } # Only configure Cloud DWH if running full integration tests diff --git a/sdk/python/tests/integration/feature_repos/universal/feature_views.py b/sdk/python/tests/integration/feature_repos/universal/feature_views.py index d2a4bab6080..ea91e9510dd 100644 --- a/sdk/python/tests/integration/feature_repos/universal/feature_views.py +++ b/sdk/python/tests/integration/feature_repos/universal/feature_views.py @@ -292,3 +292,23 @@ def create_pushable_feature_view(batch_source: DataSource): ttl=timedelta(days=2), source=push_source, ) + + +def create_vector_feature_view(source): + driver_entity = driver() + vector_tag = {"dimensions": 50} + feature_view_tag = {"index_algorithm": "hnsw"} + vector_feature_view = FeatureView( + name="driver_profile", + entities=[driver_entity], + schema= + [ + Field(name="profile_embedding", dtype=Array(base_type=Float32), tags=vector_tag), + Field(name="lifetime_trip_count", dtype=Int32), + Field(name=driver_entity.join_key, dtype=Int32), + ], + source=source, + tags=feature_view_tag, + ) + + return vector_feature_view diff --git a/sdk/python/tests/integration/online_store/test_universal_online.py b/sdk/python/tests/integration/online_store/test_universal_online.py index 82189713151..1ef3415b06a 100644 --- a/sdk/python/tests/integration/online_store/test_universal_online.py +++ b/sdk/python/tests/integration/online_store/test_universal_online.py @@ -28,6 +28,7 @@ from tests.integration.feature_repos.universal.entities import driver from tests.integration.feature_repos.universal.feature_views import ( create_driver_hourly_stats_feature_view, + create_vector_feature_view, driver_feature_view, ) from tests.utils.data_source_test_creator import prep_file_source @@ -563,6 +564,46 @@ def test_online_retrieval_success(feature_store_for_online_retrieval): ) +@pytest.mark.integration +@pytest.mark.universal_online_stores(only=["milvus"]) +def test_write_vectors_to_online_store(environment, universal_data_sources): + fs = environment.feature_store + entities, datasets, data_sources = universal_data_sources + driver_daily_stats = create_vector_feature_view(data_sources.customer) + driver_entity = driver() + + print("running apply at: " + time.strftime("%H:%M:%S", time.localtime())) + # Register Feature View and Entity + fs.apply([driver_daily_stats, driver_entity]) + + # fake data to ingest into Online Store + data = { + "driver_id": [123], + "profile_embedding": [np.random.default_rng().uniform(-100, 100, 50)], + "lifetime_trip_count": [85], + "avg_passenger_count": [0.067], + "current_balance": [0.78325], + "event_timestamp": [pd.Timestamp(datetime.datetime.utcnow()).round("ms")], + "created": [pd.Timestamp(datetime.datetime.utcnow()).round("ms")], + } + df_data = pd.DataFrame(data) + + # directly ingest data into the Online Store + fs.write_to_online_store("driver_profile", df_data) + + # assert the right data is in the Online Store + df = fs.get_online_features( + features=[ + "driver_profile:profile_embedding", + "driver_profile:lifetime_trip_count", + ], + entity_rows=[{"driver_id": 123}], + ).to_df() + assertpy.assert_that(df["profile_embedding"].iloc[0]).is_type_of(np.array) + assertpy.assert_that(df["profile_embedding"].iloc[0]).is_length(50) + assertpy.assert_that(df["lifetime_trip_count"].iloc[0]).is_equal_to(85) + + def response_feature_name( feature: str, feature_refs: List[str], full_feature_names: bool ) -> str: From dcb3a4a5fd1a64700dad76b7c72ad476262efc8e Mon Sep 17 00:00:00 2001 From: mbackes Date: Mon, 14 Aug 2023 22:50:16 -0700 Subject: [PATCH 02/12] fixed random lint issues that were raised --- sdk/python/feast/diff/registry_diff.py | 2 +- sdk/python/feast/feature_store.py | 2 +- sdk/python/feast/field.py | 2 +- sdk/python/feast/infra/offline_stores/bigquery.py | 2 +- sdk/python/feast/type_map.py | 2 +- .../tests/unit/local_feast_tests/test_local_feature_store.py | 2 +- sdk/python/tests/unit/test_feature_views.py | 2 +- 7 files changed, 7 insertions(+), 7 deletions(-) diff --git a/sdk/python/feast/diff/registry_diff.py b/sdk/python/feast/diff/registry_diff.py index 15f880e392e..2ead26131d6 100644 --- a/sdk/python/feast/diff/registry_diff.py +++ b/sdk/python/feast/diff/registry_diff.py @@ -133,7 +133,7 @@ def diff_registry_objects( if isinstance( current_proto, (DataSourceProto, ValidationReferenceProto) ) or isinstance(new_proto, (DataSourceProto, ValidationReferenceProto)): - assert type(current_proto) == type(new_proto) + assert type(current_proto) is type(new_proto) current_spec = cast(DataSourceProto, current_proto) new_spec = cast(DataSourceProto, new_proto) else: diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index ca69ab0ca98..04f0f8b5d1c 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -1109,7 +1109,7 @@ def get_historical_features( set_usage_attribute("request_fv", bool(request_feature_views)) # Check that the right request data is present in the entity_df - if type(entity_df) == pd.DataFrame: + if isinstance(entity_df, pd.DataFrame): if self.config.coerce_tz_aware: entity_df = utils.make_df_tzaware(cast(pd.DataFrame, entity_df)) for fv in request_feature_views: diff --git a/sdk/python/feast/field.py b/sdk/python/feast/field.py index c33ef03c59f..29d104eb5df 100644 --- a/sdk/python/feast/field.py +++ b/sdk/python/feast/field.py @@ -65,7 +65,7 @@ def dtype_is_feasttype_or_string_feasttype(cls, v): return v def __eq__(self, other): - if type(self) != type(other): + if type(self) is not type(other): return False if ( diff --git a/sdk/python/feast/infra/offline_stores/bigquery.py b/sdk/python/feast/infra/offline_stores/bigquery.py index 47335c411fd..632881c3622 100644 --- a/sdk/python/feast/infra/offline_stores/bigquery.py +++ b/sdk/python/feast/infra/offline_stores/bigquery.py @@ -725,7 +725,7 @@ def _get_entity_df_event_timestamp_range( or entity_df_event_timestamp_range[1] is None ): raise EntitySQLEmptyResults(entity_df) - if type(entity_df_event_timestamp_range[0]) != datetime: + if not isinstance(entity_df_event_timestamp_range[0], datetime): raise EntityDFNotDateTime() elif isinstance(entity_df, pd.DataFrame): entity_df_event_timestamp = entity_df.loc[ diff --git a/sdk/python/feast/type_map.py b/sdk/python/feast/type_map.py index df853462836..c54672b0520 100644 --- a/sdk/python/feast/type_map.py +++ b/sdk/python/feast/type_map.py @@ -745,7 +745,7 @@ def spark_to_feast_value_type(spark_type_as_str: str) -> ValueType: "array": ValueType.UNIX_TIMESTAMP_LIST, } # TODO: Find better way of doing this. - if type(spark_type_as_str) != str or spark_type_as_str not in type_map: + if not isinstance(spark_type_as_str, str) or spark_type_as_str not in type_map: return ValueType.NULL return type_map[spark_type_as_str.lower()] diff --git a/sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py b/sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py index 2cced75eb29..66a0f2ab965 100644 --- a/sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py +++ b/sdk/python/tests/unit/local_feast_tests/test_local_feature_store.py @@ -469,7 +469,7 @@ def test_apply_stream_feature_view_udf(test_feature_store, simple_dataset_1) -> def pandas_view(pandas_df): import pandas as pd - assert type(pandas_df) == pd.DataFrame + assert isinstance(pandas_df, pd.DataFrame) df = pandas_df.transform(lambda x: x + 10, axis=1) df.insert(2, "C", [20.2, 230.0, 34.0], True) return df diff --git a/sdk/python/tests/unit/test_feature_views.py b/sdk/python/tests/unit/test_feature_views.py index c954b6e53b0..f4217513009 100644 --- a/sdk/python/tests/unit/test_feature_views.py +++ b/sdk/python/tests/unit/test_feature_views.py @@ -181,7 +181,7 @@ def test_stream_feature_view_udfs(): def pandas_udf(pandas_df): import pandas as pd - assert type(pandas_df) == pd.DataFrame + assert isinstance(pandas_df, pd.DataFrame) df = pandas_df.transform(lambda x: x + 10, axis=1) return df From 2a4e64a0f4b727b12e53eeb6db04ffde7cdcc965 Mon Sep 17 00:00:00 2001 From: mbackes Date: Mon, 14 Aug 2023 22:50:39 -0700 Subject: [PATCH 03/12] formatting and cleanup --- sdk/python/feast/driver_test_data.py | 6 ++++-- .../milvus_online_store_creator.py | 21 ++++++++++++------- .../expediagroup/test_milvus_online_store.py | 12 ++++++++--- .../feature_repos/repo_configuration.py | 9 ++++---- .../feature_repos/universal/feature_views.py | 9 +++++--- 5 files changed, 37 insertions(+), 20 deletions(-) diff --git a/sdk/python/feast/driver_test_data.py b/sdk/python/feast/driver_test_data.py index 296cf07a295..21205e8f5a3 100644 --- a/sdk/python/feast/driver_test_data.py +++ b/sdk/python/feast/driver_test_data.py @@ -187,8 +187,10 @@ def create_customer_daily_profile_df(customers, start_date, end_date) -> pd.Data df_all_customers["lifetime_trip_count"] = np.random.randint( 0, 1000, size=rows ).astype(np.int32) - df_all_customers["profile_embedding"] = [np.random.default_rng().uniform(-100, 200, 50).astype(np.float32) - for _ in range(rows)] + df_all_customers["profile_embedding"] = [ + np.random.default_rng().uniform(-100, 200, 50).astype(np.float32) + for _ in range(rows) + ] # TODO: Remove created timestamp in order to test whether its really optional df_all_customers["created"] = pd.to_datetime(pd.Timestamp.now(tz=None).round("ms")) diff --git a/sdk/python/tests/expediagroup/milvus_online_store_creator.py b/sdk/python/tests/expediagroup/milvus_online_store_creator.py index b71b3750e91..ca14a87d5b0 100644 --- a/sdk/python/tests/expediagroup/milvus_online_store_creator.py +++ b/sdk/python/tests/expediagroup/milvus_online_store_creator.py @@ -1,8 +1,5 @@ -import random from typing import Dict -from milvus import default_server - from testcontainers.core.container import DockerContainer from testcontainers.core.waiting_utils import wait_for_logs @@ -14,18 +11,28 @@ class MilvusOnlineStoreCreator(OnlineStoreCreator): def __init__(self, project_name: str, **kwargs): super().__init__(project_name) - self.container = DockerContainer("milvus/milvus:2.2.12").with_exposed_ports("19530") + self.container = DockerContainer("milvus/milvus:2.2.12").with_exposed_ports( + "19530" + ) def create_online_store(self) -> Dict[str, str]: self.container.start() - log_string_to_wait_for = "Milvus Proxy successfully initialized and ready to serve!" + log_string_to_wait_for = ( + "Milvus Proxy successfully initialized and ready to serve!" + ) wait_for_logs( container=self.container, predicate=log_string_to_wait_for, timeout=20 ) exposed_port = self.container.get_exposed_port("19530") - return {"alias": "default", "type": "milvus", "host": "localhost", "port": str(exposed_port), - "username": "user", "password": "password"} + return { + "alias": "default", + "type": "milvus", + "host": "localhost", + "port": str(exposed_port), + "username": "user", + "password": "password", + } def teardown(self): self.container.stop() diff --git a/sdk/python/tests/expediagroup/test_milvus_online_store.py b/sdk/python/tests/expediagroup/test_milvus_online_store.py index a74a7eae986..d27c539b9e8 100644 --- a/sdk/python/tests/expediagroup/test_milvus_online_store.py +++ b/sdk/python/tests/expediagroup/test_milvus_online_store.py @@ -200,7 +200,9 @@ def test_milvus_update_add_collection(self, repo_config, caplog, embedded_milvus or Collection(self.collection_to_write).schema == schema2 ) - def test_milvus_update_add_existing_collection(self, repo_config, caplog, embedded_milvus): + def test_milvus_update_add_existing_collection( + self, repo_config, caplog, embedded_milvus + ): # Creating a common schema for collection feast_schema = [ Field( @@ -260,7 +262,9 @@ def test_milvus_update_add_existing_collection(self, repo_config, caplog, embedd assert utility.has_collection(self.collection_to_write) is True assert len(utility.list_collections()) == 1 - def test_milvus_update_delete_collection(self, repo_config, caplog, embedded_milvus): + def test_milvus_update_delete_collection( + self, repo_config, caplog, embedded_milvus + ): # Creating a common schema for collection which is compatible with FEAST feast_schema = [ Field( @@ -318,7 +322,9 @@ def test_milvus_update_delete_collection(self, repo_config, caplog, embedded_mil with MilvusConnectionManager(repo_config.online_store): assert utility.has_collection(self.collection_to_write) is False - def test_milvus_update_delete_unavailable_collection(self, repo_config, caplog, embedded_milvus): + def test_milvus_update_delete_unavailable_collection( + self, repo_config, caplog, embedded_milvus + ): feast_schema = [ Field( name="feature1", diff --git a/sdk/python/tests/integration/feature_repos/repo_configuration.py b/sdk/python/tests/integration/feature_repos/repo_configuration.py index 5c5d08fd895..662ed8366db 100644 --- a/sdk/python/tests/integration/feature_repos/repo_configuration.py +++ b/sdk/python/tests/integration/feature_repos/repo_configuration.py @@ -20,9 +20,7 @@ from feast.infra.feature_servers.base_config import FeatureLoggingConfig from feast.infra.feature_servers.local_process.config import LocalFeatureServerConfig from feast.repo_config import RegistryConfig, RepoConfig -from tests.expediagroup.milvus_online_store_creator import ( - MilvusOnlineStoreCreator, -) +from tests.expediagroup.milvus_online_store_creator import MilvusOnlineStoreCreator from tests.integration.feature_repos.integration_test_repo_config import ( IntegrationTestRepoConfig, RegistryLocation, @@ -108,7 +106,7 @@ "host": "localhost", "port": 19530, "username": "user", - "password": "password" + "password": "password", } OFFLINE_STORE_TO_PROVIDER_CONFIG: Dict[str, DataSourceCreator] = { @@ -126,7 +124,8 @@ str, Tuple[Union[str, Dict[str, str]], Optional[Type[OnlineStoreCreator]]] ] = { "sqlite": ({"type": "sqlite"}, None), - "milvus": (MILVUS_CONFIG, MilvusOnlineStoreCreator), + "milvus": ({"type": "milvus"}, MilvusOnlineStoreCreator), + # "milvus": (MILVUS_CONFIG, MilvusOnlineStoreCreator), } # Only configure Cloud DWH if running full integration tests diff --git a/sdk/python/tests/integration/feature_repos/universal/feature_views.py b/sdk/python/tests/integration/feature_repos/universal/feature_views.py index ea91e9510dd..7e07d1d2318 100644 --- a/sdk/python/tests/integration/feature_repos/universal/feature_views.py +++ b/sdk/python/tests/integration/feature_repos/universal/feature_views.py @@ -301,9 +301,12 @@ def create_vector_feature_view(source): vector_feature_view = FeatureView( name="driver_profile", entities=[driver_entity], - schema= - [ - Field(name="profile_embedding", dtype=Array(base_type=Float32), tags=vector_tag), + schema=[ + Field( + name="profile_embedding", + dtype=Array(base_type=Float32), + tags=vector_tag, + ), Field(name="lifetime_trip_count", dtype=Int32), Field(name=driver_entity.join_key, dtype=Int32), ], From fd887ac3323e4bfee13c8b9e2a94f44d8d33e556 Mon Sep 17 00:00:00 2001 From: mbackes Date: Mon, 14 Aug 2023 22:51:29 -0700 Subject: [PATCH 04/12] removed unused config --- .../integration/feature_repos/repo_configuration.py | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/sdk/python/tests/integration/feature_repos/repo_configuration.py b/sdk/python/tests/integration/feature_repos/repo_configuration.py index 662ed8366db..5b01f65bc4d 100644 --- a/sdk/python/tests/integration/feature_repos/repo_configuration.py +++ b/sdk/python/tests/integration/feature_repos/repo_configuration.py @@ -100,15 +100,6 @@ "host": os.getenv("ROCKSET_APISERVER", "api.rs2.usw2.rockset.com"), } -MILVUS_CONFIG = { - "alias": "default", - "type": "milvus", - "host": "localhost", - "port": 19530, - "username": "user", - "password": "password", -} - OFFLINE_STORE_TO_PROVIDER_CONFIG: Dict[str, DataSourceCreator] = { "file": ("local", FileDataSourceCreator), "bigquery": ("gcp", BigQueryDataSourceCreator), @@ -125,7 +116,6 @@ ] = { "sqlite": ({"type": "sqlite"}, None), "milvus": ({"type": "milvus"}, MilvusOnlineStoreCreator), - # "milvus": (MILVUS_CONFIG, MilvusOnlineStoreCreator), } # Only configure Cloud DWH if running full integration tests From 202f11ac69d9d2b90acccd651873f32b8908c47f Mon Sep 17 00:00:00 2001 From: mbackes Date: Mon, 14 Aug 2023 23:01:42 -0700 Subject: [PATCH 05/12] updated Milvus docker image --- sdk/python/tests/expediagroup/milvus_online_store_creator.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/python/tests/expediagroup/milvus_online_store_creator.py b/sdk/python/tests/expediagroup/milvus_online_store_creator.py index ca14a87d5b0..4eee92306b4 100644 --- a/sdk/python/tests/expediagroup/milvus_online_store_creator.py +++ b/sdk/python/tests/expediagroup/milvus_online_store_creator.py @@ -11,7 +11,7 @@ class MilvusOnlineStoreCreator(OnlineStoreCreator): def __init__(self, project_name: str, **kwargs): super().__init__(project_name) - self.container = DockerContainer("milvus/milvus:2.2.12").with_exposed_ports( + self.container = DockerContainer("mbackes/milvus-lite:2.2.12").with_exposed_ports( "19530" ) From 332de36fb707e6f43ae1974c9d56bb9d489e37e9 Mon Sep 17 00:00:00 2001 From: mbackes Date: Mon, 14 Aug 2023 23:11:31 -0700 Subject: [PATCH 06/12] reformatted and increased timeout --- .../tests/expediagroup/milvus_online_store_creator.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/sdk/python/tests/expediagroup/milvus_online_store_creator.py b/sdk/python/tests/expediagroup/milvus_online_store_creator.py index 4eee92306b4..bfc20a6bc94 100644 --- a/sdk/python/tests/expediagroup/milvus_online_store_creator.py +++ b/sdk/python/tests/expediagroup/milvus_online_store_creator.py @@ -11,9 +11,9 @@ class MilvusOnlineStoreCreator(OnlineStoreCreator): def __init__(self, project_name: str, **kwargs): super().__init__(project_name) - self.container = DockerContainer("mbackes/milvus-lite:2.2.12").with_exposed_ports( - "19530" - ) + self.container = DockerContainer( + "mbackes/milvus-lite:2.2.12" + ).with_exposed_ports("19530") def create_online_store(self) -> Dict[str, str]: self.container.start() @@ -21,7 +21,7 @@ def create_online_store(self) -> Dict[str, str]: "Milvus Proxy successfully initialized and ready to serve!" ) wait_for_logs( - container=self.container, predicate=log_string_to_wait_for, timeout=20 + container=self.container, predicate=log_string_to_wait_for, timeout=30 ) exposed_port = self.container.get_exposed_port("19530") From e33276de3ebecdb5099702b673c0220c9a16feab Mon Sep 17 00:00:00 2001 From: mbackes Date: Mon, 14 Aug 2023 23:41:47 -0700 Subject: [PATCH 07/12] Milvus unit test now using dynamically assigned port from Docker image --- .../expediagroup/test_milvus_online_store.py | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/sdk/python/tests/expediagroup/test_milvus_online_store.py b/sdk/python/tests/expediagroup/test_milvus_online_store.py index d27c539b9e8..118ace553f3 100644 --- a/sdk/python/tests/expediagroup/test_milvus_online_store.py +++ b/sdk/python/tests/expediagroup/test_milvus_online_store.py @@ -41,13 +41,14 @@ @pytest.fixture(scope="session") -def repo_config(): +def repo_config(embedded_milvus): return RepoConfig( registry=REGISTRY, project=PROJECT, provider=PROVIDER, online_store=MilvusOnlineStoreConfig( - alias=ALIAS, host=HOST, username="abc", password="cde" + alias=embedded_milvus["alias"], host=embedded_milvus["host"], port=embedded_milvus["port"], + username=embedded_milvus["username"], password=embedded_milvus["password"] ), offline_store=FileOfflineStoreConfig(), entity_key_serialization_version=2, @@ -58,9 +59,9 @@ def repo_config(): def embedded_milvus(): # Creating an online store through embedded Milvus for all tests in the class online_store_creator = MilvusOnlineStoreCreator("milvus") - online_store_creator.create_online_store() + online_store_config = online_store_creator.create_online_store() - yield online_store_creator + yield online_store_config # Tearing down the Milvus instance after all tests in the class online_store_creator.teardown() @@ -127,7 +128,7 @@ def setup_method(self, repo_config): yield - def test_milvus_update_add_collection(self, repo_config, caplog, embedded_milvus): + def test_milvus_update_add_collection(self, repo_config, caplog): feast_schema = [ Field( name="feature2", @@ -201,7 +202,7 @@ def test_milvus_update_add_collection(self, repo_config, caplog, embedded_milvus ) def test_milvus_update_add_existing_collection( - self, repo_config, caplog, embedded_milvus + self, repo_config, caplog ): # Creating a common schema for collection feast_schema = [ @@ -263,7 +264,7 @@ def test_milvus_update_add_existing_collection( assert len(utility.list_collections()) == 1 def test_milvus_update_delete_collection( - self, repo_config, caplog, embedded_milvus + self, repo_config, caplog ): # Creating a common schema for collection which is compatible with FEAST feast_schema = [ @@ -323,7 +324,7 @@ def test_milvus_update_delete_collection( assert utility.has_collection(self.collection_to_write) is False def test_milvus_update_delete_unavailable_collection( - self, repo_config, caplog, embedded_milvus + self, repo_config, caplog ): feast_schema = [ Field( From 835c87470b5f1efe3da31c9ea7c9180ce79035b1 Mon Sep 17 00:00:00 2001 From: michaelbackes Date: Mon, 14 Aug 2023 23:45:01 -0700 Subject: [PATCH 08/12] Update pr_local_integration_tests.yml disabling Milvus tests until implementation is complete --- .github/workflows/pr_local_integration_tests.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/pr_local_integration_tests.yml b/.github/workflows/pr_local_integration_tests.yml index 41df3aefff2..7b0bf3db0ae 100644 --- a/.github/workflows/pr_local_integration_tests.yml +++ b/.github/workflows/pr_local_integration_tests.yml @@ -66,4 +66,4 @@ jobs: IS_TEST: "True" FEAST_LOCAL_ONLINE_CONTAINER: "True" FEAST_IS_LOCAL_TEST: "True" - run: pytest -n 8 --cov=./ --cov-report=xml --color=yes --integration -k "not gcs_registry and not s3_registry and not test_lambda_materialization and not test_snowflake_materialization" sdk/python/tests + run: pytest -n 8 --cov=./ --cov-report=xml --color=yes --integration -k "not gcs_registry and not s3_registry and not test_lambda_materialization and not test_snowflake_materialization and not milvus" sdk/python/tests From e4b362d70da80c9498c51772f2690c380ff1f331 Mon Sep 17 00:00:00 2001 From: mbackes Date: Mon, 14 Aug 2023 23:45:48 -0700 Subject: [PATCH 09/12] formatting --- .../expediagroup/test_milvus_online_store.py | 19 ++++++++----------- 1 file changed, 8 insertions(+), 11 deletions(-) diff --git a/sdk/python/tests/expediagroup/test_milvus_online_store.py b/sdk/python/tests/expediagroup/test_milvus_online_store.py index 118ace553f3..28a7111da4e 100644 --- a/sdk/python/tests/expediagroup/test_milvus_online_store.py +++ b/sdk/python/tests/expediagroup/test_milvus_online_store.py @@ -47,8 +47,11 @@ def repo_config(embedded_milvus): project=PROJECT, provider=PROVIDER, online_store=MilvusOnlineStoreConfig( - alias=embedded_milvus["alias"], host=embedded_milvus["host"], port=embedded_milvus["port"], - username=embedded_milvus["username"], password=embedded_milvus["password"] + alias=embedded_milvus["alias"], + host=embedded_milvus["host"], + port=embedded_milvus["port"], + username=embedded_milvus["username"], + password=embedded_milvus["password"], ), offline_store=FileOfflineStoreConfig(), entity_key_serialization_version=2, @@ -201,9 +204,7 @@ def test_milvus_update_add_collection(self, repo_config, caplog): or Collection(self.collection_to_write).schema == schema2 ) - def test_milvus_update_add_existing_collection( - self, repo_config, caplog - ): + def test_milvus_update_add_existing_collection(self, repo_config, caplog): # Creating a common schema for collection feast_schema = [ Field( @@ -263,9 +264,7 @@ def test_milvus_update_add_existing_collection( assert utility.has_collection(self.collection_to_write) is True assert len(utility.list_collections()) == 1 - def test_milvus_update_delete_collection( - self, repo_config, caplog - ): + def test_milvus_update_delete_collection(self, repo_config, caplog): # Creating a common schema for collection which is compatible with FEAST feast_schema = [ Field( @@ -323,9 +322,7 @@ def test_milvus_update_delete_collection( with MilvusConnectionManager(repo_config.online_store): assert utility.has_collection(self.collection_to_write) is False - def test_milvus_update_delete_unavailable_collection( - self, repo_config, caplog - ): + def test_milvus_update_delete_unavailable_collection(self, repo_config, caplog): feast_schema = [ Field( name="feature1", From 8026b9a0e1e8873a9fe22f6663b8a008220395a5 Mon Sep 17 00:00:00 2001 From: michaelbackes Date: Tue, 15 Aug 2023 00:02:45 -0700 Subject: [PATCH 10/12] Update pr_local_integration_tests.yml reverting previous change --- .github/workflows/pr_local_integration_tests.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/pr_local_integration_tests.yml b/.github/workflows/pr_local_integration_tests.yml index 7b0bf3db0ae..41df3aefff2 100644 --- a/.github/workflows/pr_local_integration_tests.yml +++ b/.github/workflows/pr_local_integration_tests.yml @@ -66,4 +66,4 @@ jobs: IS_TEST: "True" FEAST_LOCAL_ONLINE_CONTAINER: "True" FEAST_IS_LOCAL_TEST: "True" - run: pytest -n 8 --cov=./ --cov-report=xml --color=yes --integration -k "not gcs_registry and not s3_registry and not test_lambda_materialization and not test_snowflake_materialization and not milvus" sdk/python/tests + run: pytest -n 8 --cov=./ --cov-report=xml --color=yes --integration -k "not gcs_registry and not s3_registry and not test_lambda_materialization and not test_snowflake_materialization" sdk/python/tests From df16f12c7593bdf6d82275e9d4b16bb989161e3b Mon Sep 17 00:00:00 2001 From: mbackes Date: Tue, 15 Aug 2023 00:15:30 -0700 Subject: [PATCH 11/12] disabling Milvus integration tests until implementation is complete --- .../tests/integration/feature_repos/repo_configuration.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sdk/python/tests/integration/feature_repos/repo_configuration.py b/sdk/python/tests/integration/feature_repos/repo_configuration.py index 5b01f65bc4d..6458bf362ea 100644 --- a/sdk/python/tests/integration/feature_repos/repo_configuration.py +++ b/sdk/python/tests/integration/feature_repos/repo_configuration.py @@ -20,7 +20,6 @@ from feast.infra.feature_servers.base_config import FeatureLoggingConfig from feast.infra.feature_servers.local_process.config import LocalFeatureServerConfig from feast.repo_config import RegistryConfig, RepoConfig -from tests.expediagroup.milvus_online_store_creator import MilvusOnlineStoreCreator from tests.integration.feature_repos.integration_test_repo_config import ( IntegrationTestRepoConfig, RegistryLocation, @@ -115,7 +114,8 @@ str, Tuple[Union[str, Dict[str, str]], Optional[Type[OnlineStoreCreator]]] ] = { "sqlite": ({"type": "sqlite"}, None), - "milvus": ({"type": "milvus"}, MilvusOnlineStoreCreator), + # uncomment below once Milvus implementation is complete + # "milvus": ({"type": "milvus"}, MilvusOnlineStoreCreator), } # Only configure Cloud DWH if running full integration tests From 24807f4f3d27ed50c5b6ce5467634a8b2f6c8aad Mon Sep 17 00:00:00 2001 From: michaelbackes Date: Tue, 15 Aug 2023 01:53:19 -0700 Subject: [PATCH 12/12] Update unit_tests.yml need to exclude unit-tests for mac os - there is an issue with the Milvus Docker image that needs to be addressed for them to work --- .github/workflows/unit_tests.yml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.github/workflows/unit_tests.yml b/.github/workflows/unit_tests.yml index 6425171df39..5cb2eae7111 100644 --- a/.github/workflows/unit_tests.yml +++ b/.github/workflows/unit_tests.yml @@ -10,6 +10,8 @@ jobs: python-version: ["3.8", "3.9", "3.10"] os: [ubuntu-latest, macOS-latest] exclude: + - os: macOS-latest + python-version: "3.8" - os: macOS-latest python-version: "3.9" - os: macOS-latest