fixing a bad merge/rebase

Signed-off-by: Francisco Javier Arceo <farceo@redhat.com>
franciscojavierarceo · Jan 30, 2025 · Jan 25, 2025 · Jan 27, 2025 · Jan 27, 2025 · Jan 8, 2025
commit 63aa43bffd013ba319291a89b4a996c6b6bb307e
diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py
@@ -866,7 +866,8 @@ def apply(
         views_to_update = [
             ob
             for ob in objects
-            if (
+            if
+            (
                 # BFVs are not handled separately from FVs right now.
                 (isinstance(ob, FeatureView) or isinstance(ob, BatchFeatureView))
                 and not isinstance(ob, StreamFeatureView)
@@ -1848,7 +1849,9 @@ def retrieve_online_documents(
                     join_key_values[join_key].append(entity_value)
 
             document_feature_vals = [feature[4] for feature in document_features]
-            document_feature_distance_vals = [feature[5] for feature in document_features]
+            document_feature_distance_vals = [
+                feature[5] for feature in document_features
+            ]
             online_features_response = GetOnlineFeaturesResponse(results=[])
             requested_feature = requested_feature or requested_features[0]
             utils._populate_result_rows_from_columnar(
@@ -1864,7 +1867,6 @@ def retrieve_online_documents(
             return self._retrieve_from_online_store_v2(
                 provider,
                 requested_feature_view,
-                requested_feature,
                 requested_features,
                 query,
                 top_k,
@@ -2168,9 +2170,9 @@ def write_logged_features(
         if not isinstance(source, FeatureService):
             raise ValueError("Only feature service is currently supported as a source")
 
-        assert source.logging_config is not None, (
-            "Feature service must be configured with logging config in order to use this functionality"
-        )
+        assert (
+            source.logging_config is not None
+        ), "Feature service must be configured with logging config in order to use this functionality"
 
         assert isinstance(logs, (pa.Table, Path))
 

diff --git a/sdk/python/feast/infra/materialization/snowflake_engine.py b/sdk/python/feast/infra/materialization/snowflake_engine.py
@@ -206,9 +206,9 @@ def __init__(
         online_store: OnlineStore,
         **kwargs,
     ):
-        assert repo_config.offline_store.type == "snowflake.offline", (
-            "To use SnowflakeMaterializationEngine, you must use Snowflake as an offline store."
-        )
+        assert (
+            repo_config.offline_store.type == "snowflake.offline"
+        ), "To use SnowflakeMaterializationEngine, you must use Snowflake as an offline store."
 
         super().__init__(
             repo_config=repo_config,
@@ -241,11 +241,10 @@ def _materialize_one(
         project: str,
         tqdm_builder: Callable[[int], tqdm],
     ):
-        assert isinstance(feature_view, BatchFeatureView) or isinstance(
-            feature_view, FeatureView
-        ), (
-            "Snowflake can only materialize FeatureView & BatchFeatureView feature view types."
-        )
+        assert (
+            isinstance(feature_view, BatchFeatureView)
+            or isinstance(feature_view, FeatureView)
+        ), "Snowflake can only materialize FeatureView & BatchFeatureView feature view types."
 
         entities = []
         for entity_name in feature_view.entities:

diff --git a/sdk/python/feast/infra/offline_stores/dask.py b/sdk/python/feast/infra/offline_stores/dask.py
@@ -193,7 +193,9 @@ def evaluate_historical_retrieval():
             ):
                 # Make sure all event timestamp fields are tz-aware. We default tz-naive fields to UTC
                 entity_df_with_features[entity_df_event_timestamp_col] = (
-                    entity_df_with_features[entity_df_event_timestamp_col].apply(
+                    entity_df_with_features[
+                        entity_df_event_timestamp_col
+                    ].apply(
                         lambda x: x
                         if x.tzinfo is not None
                         else x.replace(tzinfo=timezone.utc)

@@ -375,9 +375,7 @@ def retrieve_online_documents_v2(
         assert all(
             field in [f["name"] for f in collection["fields"]]
             for field in output_fields
-        ), (
-            f"field(s) [{[field for field in output_fields if field not in [f['name'] for f in collection['fields']]]}] not found in collection schema"
-        )
+        ), f"field(s) [{[field for field in output_fields if field not in [f['name'] for f in collection['fields']]]}] not found in collection schema"
         # Note we choose the first vector field as the field to search on. Not ideal but it's something.
         ann_search_field = None
         for field in collection["fields"]:

diff --git a/sdk/python/feast/infra/online_stores/qdrant_online_store/qdrant.py b/sdk/python/feast/infra/online_stores/qdrant_online_store/qdrant.py
@@ -69,9 +69,9 @@ def _get_client(self, config: RepoConfig) -> QdrantClient:
         if self._client:
             return self._client
         online_store_config = config.online_store
-        assert isinstance(online_store_config, QdrantOnlineStoreConfig), (
-            "Invalid type for online store config"
-        )
+        assert isinstance(
+            online_store_config, QdrantOnlineStoreConfig
+        ), "Invalid type for online store config"
 
         assert online_store_config.similarity and (
             online_store_config.similarity.lower() in DISTANCE_MAPPING

diff --git a/sdk/python/feast/infra/passthrough_provider.py b/sdk/python/feast/infra/passthrough_provider.py
@@ -494,9 +494,9 @@ def write_feature_service_logs(
         config: RepoConfig,
         registry: BaseRegistry,
     ):
-        assert feature_service.logging_config is not None, (
-            "Logging should be configured for the feature service before calling this function"
-        )
+        assert (
+            feature_service.logging_config is not None
+        ), "Logging should be configured for the feature service before calling this function"
 
         self.offline_store.write_logged_features(
             config=config,
@@ -514,9 +514,9 @@ def retrieve_feature_service_logs(
         config: RepoConfig,
         registry: BaseRegistry,
     ) -> RetrievalJob:
-        assert feature_service.logging_config is not None, (
-            "Logging should be configured for the feature service before calling this function"
-        )
+        assert (
+            feature_service.logging_config is not None
+        ), "Logging should be configured for the feature service before calling this function"
 
         logging_source = FeatureServiceLoggingSource(feature_service, config.project)
         schema = logging_source.get_schema(registry)

diff --git a/sdk/python/feast/offline_server.py b/sdk/python/feast/offline_server.py
@@ -266,15 +266,15 @@ def do_get(self, context: fl.ServerCallContext, ticket: fl.Ticket):
         return fl.RecordBatchStream(table)
 
     def _validate_offline_write_batch_parameters(self, command: dict):
-        assert "feature_view_names" in command, (
-            "feature_view_names is a mandatory parameter"
-        )
+        assert (
+            "feature_view_names" in command
+        ), "feature_view_names is a mandatory parameter"
         assert "name_aliases" in command, "name_aliases is a mandatory parameter"
 
         feature_view_names = command["feature_view_names"]
-        assert len(feature_view_names) == 1, (
-            "feature_view_names list should only have one item"
-        )
+        assert (
+            len(feature_view_names) == 1
+        ), "feature_view_names list should only have one item"
 
         name_aliases = command["name_aliases"]
         assert len(name_aliases) == 1, "name_aliases list should only have one item"
@@ -316,9 +316,9 @@ def write_logged_features(self, command: dict, key: str):
             command["feature_service_name"]
         )
 
-        assert feature_service.logging_config is not None, (
-            "feature service must have logging_config set"
-        )
+        assert (
+            feature_service.logging_config is not None
+        ), "feature service must have logging_config set"
 
         assert_permissions(
             resource=feature_service,
@@ -335,15 +335,15 @@ def write_logged_features(self, command: dict, key: str):
         )
 
     def _validate_pull_all_from_table_or_query_parameters(self, command: dict):
-        assert "data_source_name" in command, (
-            "data_source_name is a mandatory parameter"
-        )
-        assert "join_key_columns" in command, (
-            "join_key_columns is a mandatory parameter"
-        )
-        assert "feature_name_columns" in command, (
-            "feature_name_columns is a mandatory parameter"
-        )
+        assert (
+            "data_source_name" in command
+        ), "data_source_name is a mandatory parameter"
+        assert (
+            "join_key_columns" in command
+        ), "join_key_columns is a mandatory parameter"
+        assert (
+            "feature_name_columns" in command
+        ), "feature_name_columns is a mandatory parameter"
         assert "timestamp_field" in command, "timestamp_field is a mandatory parameter"
         assert "start_date" in command, "start_date is a mandatory parameter"
         assert "end_date" in command, "end_date is a mandatory parameter"
@@ -364,15 +364,15 @@ def pull_all_from_table_or_query(self, command: dict):
         )
 
     def _validate_pull_latest_from_table_or_query_parameters(self, command: dict):
-        assert "data_source_name" in command, (
-            "data_source_name is a mandatory parameter"
-        )
-        assert "join_key_columns" in command, (
-            "join_key_columns is a mandatory parameter"
-        )
-        assert "feature_name_columns" in command, (
-            "feature_name_columns is a mandatory parameter"
-        )
+        assert (
+            "data_source_name" in command
+        ), "data_source_name is a mandatory parameter"
+        assert (
+            "join_key_columns" in command
+        ), "join_key_columns is a mandatory parameter"
+        assert (
+            "feature_name_columns" in command
+        ), "feature_name_columns is a mandatory parameter"
         assert "timestamp_field" in command, "timestamp_field is a mandatory parameter"
         assert "start_date" in command, "start_date is a mandatory parameter"
         assert "end_date" in command, "end_date is a mandatory parameter"

diff --git a/sdk/python/feast/type_map.py b/sdk/python/feast/type_map.py
@@ -458,13 +458,13 @@ def _python_value_to_proto_value(
                 # Numpy convert 0 to int. However, in the feature view definition, the type of column may be a float.
                 # So, if value is 0, type validation must pass if scalar_types are either int or float.
                 allowed_types = {np.int64, int, np.float64, float}
-                assert type(sample) in allowed_types, (
-                    f"Type `{type(sample)}` not in {allowed_types}"
-                )
+                assert (
+                    type(sample) in allowed_types
+                ), f"Type `{type(sample)}` not in {allowed_types}"
             else:
-                assert type(sample) in valid_scalar_types, (
-                    f"Type `{type(sample)}` not in {valid_scalar_types}"
-                )
+                assert (
+                    type(sample) in valid_scalar_types
+                ), f"Type `{type(sample)}` not in {valid_scalar_types}"
         if feast_value_type == ValueType.BOOL:
             # ProtoValue does not support conversion of np.bool_ so we need to convert it to support np.bool_.
             return [

diff --git a/sdk/python/feast/types.py b/sdk/python/feast/types.py
@@ -207,9 +207,9 @@ def from_feast_to_pyarrow_type(feast_type: FeastType) -> pyarrow.DataType:
     Raises:
         ValueError: The conversion could not be performed.
     """
-    assert isinstance(feast_type, (ComplexFeastType, PrimitiveFeastType)), (
-        f"Expected FeastType, got {type(feast_type)}"
-    )
+    assert isinstance(
+        feast_type, (ComplexFeastType, PrimitiveFeastType)
+    ), f"Expected FeastType, got {type(feast_type)}"
     if isinstance(feast_type, PrimitiveFeastType):
         if feast_type in FEAST_TYPES_TO_PYARROW_TYPES:
             return FEAST_TYPES_TO_PYARROW_TYPES[feast_type]

diff --git a/sdk/python/feast/utils.py b/sdk/python/feast/utils.py
@@ -738,35 +738,6 @@ def _get_unique_entities_from_values(
     return unique_entities, indexes
 
 
-def _get_unique_entities_from_values(
-    table_entity_values: Dict[str, List[ValueProto]],
-) -> Tuple[Tuple[Dict[str, ValueProto], ...], Tuple[List[int], ...]]:
-    """Return the set of unique composite Entities for a Feature View and the indexes at which they appear.
-
-    This method allows us to query the OnlineStore for data we need only once
-    rather than requesting and processing data for the same combination of
-    Entities multiple times.
-    """
-    keys = table_entity_values.keys()
-    # Sort the rowise data to allow for grouping but keep original index. This lambda is
-    # sufficient as Entity types cannot be complex (ie. lists).
-    rowise = list(enumerate(zip(*table_entity_values.values())))
-    rowise.sort(key=lambda row: tuple(getattr(x, x.WhichOneof("val")) for x in row[1]))
-
-    # Identify unique entities and the indexes at which they occur.
-    unique_entities: Tuple[Dict[str, ValueProto], ...]
-    indexes: Tuple[List[int], ...]
-    unique_entities, indexes = tuple(
-        zip(
-            *[
-                (dict(zip(keys, k)), [_[0] for _ in g])
-                for k, g in itertools.groupby(rowise, key=lambda x: x[1])
-            ]
-        )
-    )
-    return unique_entities, indexes
-
-
 def _drop_unneeded_columns(
     online_features_response: GetOnlineFeaturesResponse,
     requested_result_row_names: Set[str],
@@ -936,69 +907,6 @@ def _populate_response_from_feature_data_v2(
         )
 
 
-def _convert_entity_key_to_proto_to_dict(
-    entity_key_vals: List[EntityKeyProto],
-) -> Dict[str, List[ValueProto]]:
-    entity_dict: Dict[str, List[ValueProto]] = {}
-    for entity_key_val in entity_key_vals:
-        if entity_key_val is not None:
-            for join_key, entity_value in zip(
-                entity_key_val.join_keys, entity_key_val.entity_values
-            ):
-                if join_key not in entity_dict:
-                    entity_dict[join_key] = []
-                    # python_entity_value = _proto_value_to_value_type(entity_value)
-                    entity_dict[join_key].append(entity_value)
-    return entity_dict
-
-def _populate_response_from_feature_data_v2(
-    feature_data: Iterable[
-        Tuple[
-            Iterable[Timestamp], Iterable["FieldStatus.ValueType"], Iterable[ValueProto]
-        ]
-    ],
-    indexes: Iterable[List[int]],
-    online_features_response: GetOnlineFeaturesResponse,
-    requested_features: Iterable[str],
-):
-    """Populate the GetOnlineFeaturesResponse with feature data.
-
-    This method assumes that `_read_from_online_store` returns data for each
-    combination of Entities in `entity_rows` in the same order as they
-    are provided.
-
-    Args:
-        feature_data: A list of data in Protobuf form which was retrieved from the OnlineStore.
-        indexes: A list of indexes which should be the same length as `feature_data`. Each list
-            of indexes corresponds to a set of result rows in `online_features_response`.
-        online_features_response: The object to populate.
-        full_feature_names: A boolean that provides the option to add the feature view prefixes to the feature names,
-            changing them from the format "feature" to "feature_view__feature" (e.g., "daily_transactions" changes to
-            "customer_fv__daily_transactions").
-        requested_features: The names of the features in `feature_data`. This should be ordered in the same way as the
-            data in `feature_data`.
-    """
-    # Add the feature names to the response.
-    requested_feature_refs = [(feature_name) for feature_name in requested_features]
-    online_features_response.metadata.feature_names.val.extend(requested_feature_refs)
-
-    timestamps, statuses, values = zip(*feature_data)
-
-    # Populate the result with data fetched from the OnlineStore
-    # which is guaranteed to be aligned with `requested_features`.
-    for (
-        feature_idx,
-        (timestamp_vector, statuses_vector, values_vector),
-    ) in enumerate(zip(zip(*timestamps), zip(*statuses), zip(*values))):
-        online_features_response.results.append(
-            GetOnlineFeaturesResponse.FeatureVector(
-                values=apply_list_mapping(values_vector, indexes),
-                statuses=apply_list_mapping(statuses_vector, indexes),
-                event_timestamps=apply_list_mapping(timestamp_vector, indexes),
-            )
-        )
-
-
 def _convert_entity_key_to_proto_to_dict(
     entity_key_vals: List[EntityKeyProto],
 ) -> Dict[str, List[ValueProto]]:

diff --git a/sdk/python/tests/integration/materialization/test_snowflake.py b/sdk/python/tests/integration/materialization/test_snowflake.py
@@ -178,9 +178,9 @@ def test_snowflake_materialization_consistency_internal_with_lists(
         assert actual_value is not None, f"Response: {response_dict}"
         if feature_dtype == "float":
             for actual_num, expected_num in zip(actual_value, expected_value):
-                assert abs(actual_num - expected_num) < 1e-6, (
-                    f"Response: {response_dict}, Expected: {expected_value}"
-                )
+                assert (
+                    abs(actual_num - expected_num) < 1e-6
+                ), f"Response: {response_dict}, Expected: {expected_value}"
         else:
             assert actual_value == expected_value
 

diff --git a/sdk/python/tests/integration/registration/test_universal_types.py b/sdk/python/tests/integration/registration/test_universal_types.py
@@ -171,9 +171,9 @@ def test_feature_get_online_features_types_match(
     if config.feature_is_list:
         for feature in online_features["value"]:
             assert isinstance(feature, list), "Feature value should be a list"
-            assert config.has_empty_list or len(feature) > 0, (
-                "List of values should not be empty"
-            )
+            assert (
+                config.has_empty_list or len(feature) > 0
+            ), "List of values should not be empty"
             for element in feature:
                 assert isinstance(element, expected_dtype)
     else:
@@ -224,9 +224,7 @@ def assert_expected_historical_feature_types(
     dtype_checkers = feature_dtype_to_expected_historical_feature_dtype[feature_dtype]
     assert any(
         check(historical_features_df.dtypes["value"]) for check in dtype_checkers
-    ), (
-        f"Failed to match feature type {historical_features_df.dtypes['value']} with checkers {dtype_checkers}"
-    )
+    ), f"Failed to match feature type {historical_features_df.dtypes['value']} with checkers {dtype_checkers}"
 
 
 def assert_feature_list_types(