chore: Suppress alpha warnings in test. Fix entity serialization in t…

…est (#3029)

* chore: Suppress alpha warnings in test. Fix entity serialization in test

Signed-off-by: Danny Chiao <danny@tecton.ai>

* fix typo

Signed-off-by: Danny Chiao <danny@tecton.ai>

* revert lambda test

Signed-off-by: Danny Chiao <danny@tecton.ai>

.github/workflows/pr_local_integration_tests.yml

@@ -15,7 +15,7 @@ jobs:
       ((github.event.action == 'labeled' && (github.event.label.name == 'approved' || github.event.label.name == 'lgtm' || github.event.label.name == 'ok-to-test')) ||
       (github.event.action != 'labeled' && (contains(github.event.pull_request.labels.*.name, 'ok-to-test') || contains(github.event.pull_request.labels.*.name, 'approved') || contains(github.event.pull_request.labels.*.name, 'lgtm')))) ||
       github.repository != 'feast-dev/feast'
-    runs-on: ${{ matrix.os }}p
+    runs-on: ${{ matrix.os }}
     strategy:
       fail-fast: false
       matrix:

protos/feast/core/StreamFeatureView.proto

@@ -24,7 +24,6 @@ option java_package = "feast.proto.core";
 
 
 import "google/protobuf/duration.proto";
-import "google/protobuf/timestamp.proto";
 import "feast/core/OnDemandFeatureView.proto";
 import "feast/core/FeatureView.proto";
 import "feast/core/Feature.proto";

sdk/python/feast/driver_test_data.py

@@ -164,7 +164,7 @@ def create_customer_daily_profile_df(customers, start_date, end_date) -> pd.Data
             "event_timestamp": [
                 pd.Timestamp(dt, unit="ms", tz="UTC").round("ms")
                 for dt in pd.date_range(
-                    start=start_date, end=end_date, freq="1D", closed="left"
+                    start=start_date, end=end_date, freq="1D", inclusive="left"
                 )
             ]
         }
@@ -209,7 +209,7 @@ def create_location_stats_df(locations, start_date, end_date) -> pd.DataFrame:
             "event_timestamp": [
                 pd.Timestamp(dt, unit="ms", tz="UTC").round("ms")
                 for dt in pd.date_range(
-                    start=start_date, end=end_date, freq="1H", closed="left"
+                    start=start_date, end=end_date, freq="1H", inclusive="left"
                 )
             ]
         }
@@ -256,7 +256,7 @@ def create_global_daily_stats_df(start_date, end_date) -> pd.DataFrame:
             "event_timestamp": [
                 pd.Timestamp(dt, unit="ms", tz="UTC").round("ms")
                 for dt in pd.date_range(
-                    start=start_date, end=end_date, freq="1D", closed="left"
+                    start=start_date, end=end_date, freq="1D", inclusive="left"
                 )
             ]
         }

sdk/python/feast/feature_store.py

@@ -40,7 +40,7 @@
 from google.protobuf.timestamp_pb2 import Timestamp
 from tqdm import tqdm
 
-from feast import feature_server, ui_server, utils
+from feast import feature_server, flags_helper, ui_server, utils
 from feast.base_feature_view import BaseFeatureView
 from feast.batch_feature_view import BatchFeatureView
 from feast.data_source import DataSource, PushMode
@@ -533,7 +533,7 @@ def _validate_all_feature_views(
         sfvs_to_update: List[StreamFeatureView],
     ):
         """Validates all feature views."""
-        if len(odfvs_to_update) > 0:
+        if len(odfvs_to_update) > 0 and not flags_helper.is_test():
             warnings.warn(
                 "On demand feature view is an experimental feature. "
                 "This API is stable, but the functionality does not scale well for offline retrieval",
@@ -1123,12 +1123,13 @@ def create_saved_dataset(
         Raises:
             ValueError if given retrieval job doesn't have metadata
         """
-        warnings.warn(
-            "Saving dataset is an experimental feature. "
-            "This API is unstable and it could and most probably will be changed in the future. "
-            "We do not guarantee that future changes will maintain backward compatibility.",
-            RuntimeWarning,
-        )
+        if not flags_helper.is_test():
+            warnings.warn(
+                "Saving dataset is an experimental feature. "
+                "This API is unstable and it could and most probably will be changed in the future. "
+                "We do not guarantee that future changes will maintain backward compatibility.",
+                RuntimeWarning,
+            )
 
         if not from_.metadata:
             raise ValueError(
@@ -1175,12 +1176,13 @@ def get_saved_dataset(self, name: str) -> SavedDataset:
         Raises:
             SavedDatasetNotFound
         """
-        warnings.warn(
-            "Retrieving datasets is an experimental feature. "
-            "This API is unstable and it could and most probably will be changed in the future. "
-            "We do not guarantee that future changes will maintain backward compatibility.",
-            RuntimeWarning,
-        )
+        if not flags_helper.is_test():
+            warnings.warn(
+                "Retrieving datasets is an experimental feature. "
+                "This API is unstable and it could and most probably will be changed in the future. "
+                "We do not guarantee that future changes will maintain backward compatibility.",
+                RuntimeWarning,
+            )
 
         dataset = self._registry.get_saved_dataset(name, self.project)
         provider = self._get_provider()
@@ -1374,12 +1376,13 @@ def push(
             allow_registry_cache: Whether to allow cached versions of the registry.
             to: Whether to push to online or offline store. Defaults to online store only.
         """
-        warnings.warn(
-            "Push source is an experimental feature. "
-            "This API is unstable and it could and might change in the future. "
-            "We do not guarantee that future changes will maintain backward compatibility.",
-            RuntimeWarning,
-        )
+        if not flags_helper.is_test():
+            warnings.warn(
+                "Push source is an experimental feature. "
+                "This API is unstable and it could and might change in the future. "
+                "We do not guarantee that future changes will maintain backward compatibility.",
+                RuntimeWarning,
+            )
         from feast.data_source import PushSource
 
         all_fvs = self.list_feature_views(allow_cache=allow_registry_cache)
@@ -2268,11 +2271,12 @@ def serve_ui(
         self, host: str, port: int, get_registry_dump: Callable, registry_ttl_sec: int
     ) -> None:
         """Start the UI server locally"""
-        warnings.warn(
-            "The Feast UI is an experimental feature. "
-            "We do not guarantee that future changes will maintain backward compatibility.",
-            RuntimeWarning,
-        )
+        if flags_helper.is_test():
+            warnings.warn(
+                "The Feast UI is an experimental feature. "
+                "We do not guarantee that future changes will maintain backward compatibility.",
+                RuntimeWarning,
+            )
         ui_server.start_server(
             self,
             host=host,
@@ -2352,12 +2356,13 @@ def validate_logged_features(
             or None if successful.
 
         """
-        warnings.warn(
-            "Logged features validation is an experimental feature. "
-            "This API is unstable and it could and most probably will be changed in the future. "
-            "We do not guarantee that future changes will maintain backward compatibility.",
-            RuntimeWarning,
-        )
+        if not flags_helper.is_test():
+            warnings.warn(
+                "Logged features validation is an experimental feature. "
+                "This API is unstable and it could and most probably will be changed in the future. "
+                "We do not guarantee that future changes will maintain backward compatibility.",
+                RuntimeWarning,
+            )
 
         if not isinstance(source, FeatureService):
             raise ValueError("Only feature service is currently supported as a source")

sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark_source.py

@@ -6,6 +6,7 @@
 
 from pyspark.sql import SparkSession
 
+from feast import flags_helper
 from feast.data_source import DataSource
 from feast.errors import DataSourceNoNameException
 from feast.infra.offline_stores.offline_utils import get_temp_entity_table_name
@@ -62,11 +63,12 @@ def __init__(
             owner=owner,
         )
 
-        warnings.warn(
-            "The spark data source API is an experimental feature in alpha development. "
-            "This API is unstable and it could and most probably will be changed in the future.",
-            RuntimeWarning,
-        )
+        if not flags_helper.is_test():
+            warnings.warn(
+                "The spark data source API is an experimental feature in alpha development. "
+                "This API is unstable and it could and most probably will be changed in the future.",
+                RuntimeWarning,
+            )
 
         self.spark_options = SparkOptions(
             table=table,

sdk/python/feast/infra/offline_stores/offline_store.py

@@ -20,6 +20,7 @@
 import pandas as pd
 import pyarrow
 
+from feast import flags_helper
 from feast.data_source import DataSource
 from feast.dqm.errors import ValidationFailed
 from feast.feature_logging import LoggingConfig, LoggingSource
@@ -91,12 +92,13 @@ def to_df(
                 )
 
         if validation_reference:
-            warnings.warn(
-                "Dataset validation is an experimental feature. "
-                "This API is unstable and it could and most probably will be changed in the future. "
-                "We do not guarantee that future changes will maintain backward compatibility.",
-                RuntimeWarning,
-            )
+            if not flags_helper.is_test():
+                warnings.warn(
+                    "Dataset validation is an experimental feature. "
+                    "This API is unstable and it could and most probably will be changed in the future. "
+                    "We do not guarantee that future changes will maintain backward compatibility.",
+                    RuntimeWarning,
+                )
 
             validation_result = validation_reference.profile.validate(features_df)
             if not validation_result.is_success:
@@ -136,12 +138,13 @@ def to_arrow(
                 )
 
         if validation_reference:
-            warnings.warn(
-                "Dataset validation is an experimental feature. "
-                "This API is unstable and it could and most probably will be changed in the future. "
-                "We do not guarantee that future changes will maintain backward compatibility.",
-                RuntimeWarning,
-            )
+            if not flags_helper.is_test():
+                warnings.warn(
+                    "Dataset validation is an experimental feature. "
+                    "This API is unstable and it could and most probably will be changed in the future. "
+                    "We do not guarantee that future changes will maintain backward compatibility.",
+                    RuntimeWarning,
+                )
 
             validation_result = validation_reference.profile.validate(features_df)
             if not validation_result.is_success:

sdk/python/feast/stream_feature_view.py

@@ -8,7 +8,7 @@
 import dill
 from typeguard import typechecked
 
-from feast import utils
+from feast import flags_helper, utils
 from feast.aggregation import Aggregation
 from feast.data_source import DataSource
 from feast.entity import Entity
@@ -90,11 +90,12 @@ def __init__(
         timestamp_field: Optional[str] = "",
         udf: Optional[FunctionType] = None,
     ):
-        warnings.warn(
-            "Stream Feature Views are experimental features in alpha development. "
-            "Some functionality may still be unstable so functionality can change in the future.",
-            RuntimeWarning,
-        )
+        if not flags_helper.is_test():
+            warnings.warn(
+                "Stream Feature Views are experimental features in alpha development. "
+                "Some functionality may still be unstable so functionality can change in the future.",
+                RuntimeWarning,
+            )
 
         if (
             type(source).__name__ not in SUPPORTED_STREAM_SOURCES

sdk/python/tests/integration/e2e/test_usage_e2e.py

@@ -57,6 +57,7 @@ def test_usage_on(dummy_exporter, enabling_toggle):
                 online_store=SqliteOnlineStoreConfig(
                     path=os.path.join(temp_dir, "online.db")
                 ),
+                entity_key_serialization_version=2,
             )
         )
         entity = Entity(
@@ -95,6 +96,7 @@ def test_usage_off(dummy_exporter, enabling_toggle):
                 online_store=SqliteOnlineStoreConfig(
                     path=os.path.join(temp_dir, "online.db")
                 ),
+                entity_key_serialization_version=2,
             )
         )
         entity = Entity(

sdk/python/tests/integration/e2e/test_validation.py

@@ -150,25 +150,32 @@ def test_logged_features_validation(environment, universal_data_sources):
 
     # add some non-existing entities to check NotFound feature handling
     for i in range(5):
-        entity_df = entity_df.append(
-            {
-                "customer_id": 2000 + i,
-                "driver_id": 6000 + i,
-                "event_timestamp": datetime.datetime.now(),
-            },
-            ignore_index=True,
+        entity_df = pd.concat(
+            [
+                entity_df,
+                pd.DataFrame.from_records(
+                    [
+                        {
+                            "customer_id": 2000 + i,
+                            "driver_id": 6000 + i,
+                            "event_timestamp": datetime.datetime.now(),
+                        }
+                    ]
+                ),
+            ]
         )
 
+    store_fs = store.get_feature_service(feature_service.name)
     reference_dataset = store.create_saved_dataset(
         from_=store.get_historical_features(
-            entity_df=entity_df, features=feature_service, full_feature_names=True
+            entity_df=entity_df, features=store_fs, full_feature_names=True
         ),
         name="reference_for_validating_logged_features",
         storage=environment.data_source_creator.create_saved_dataset_destination(),
     )
 
     log_source_df = store.get_historical_features(
-        entity_df=entity_df, features=feature_service, full_feature_names=False
+        entity_df=entity_df, features=store_fs, full_feature_names=False
     ).to_df()
     logs_df = prepare_logs(log_source_df, feature_service, store)
 
@@ -229,7 +236,9 @@ def test_e2e_validation_via_cli(environment, universal_data_sources):
         columns=["order_id", "origin_id", "destination_id", "driver_id"]
     )
     retrieval_job = store.get_historical_features(
-        entity_df=entity_df, features=feature_service, full_feature_names=True
+        entity_df=entity_df,
+        features=store.get_feature_service(feature_service.name),
+        full_feature_names=True,
     )
     logs_df = prepare_logs(retrieval_job.to_df(), feature_service, store)
     saved_dataset = store.create_saved_dataset(

sdk/python/tests/integration/feature_repos/repo_configuration.py

@@ -368,6 +368,7 @@ def construct_test_environment(
     fixture_request: Optional[pytest.FixtureRequest],
     test_suite_name: str = "integration_test",
     worker_id: str = "worker_id",
+    entity_key_serialization_version: int = 2,
 ) -> Environment:
     _uuid = str(uuid.uuid4()).replace("-", "")[:6]
 
@@ -437,6 +438,7 @@ def construct_test_environment(
         repo_path=repo_dir_name,
         feature_server=feature_server,
         go_feature_serving=test_repo_config.go_feature_serving,
+        entity_key_serialization_version=entity_key_serialization_version,
     )
 
     # Create feature_store.yaml out of the config

sdk/python/tests/integration/materialization/test_lambda.py

@@ -33,7 +33,10 @@ def test_lambda_materialization_consistency():
         },
         registry_location=RegistryLocation.S3,
     )
-    lambda_environment = construct_test_environment(lambda_config, None)
+    # TODO(adchia): figure out why entity_key_serialization_version 2 breaks with this test
+    lambda_environment = construct_test_environment(
+        lambda_config, None, entity_key_serialization_version=1
+    )
 
     df = create_basic_driver_dataset()
     ds = lambda_environment.data_source_creator.create_data_source(

sdk/python/tests/integration/registration/test_feature_store.py

@@ -194,6 +194,7 @@ def feature_store_with_local_registry():
             project="default",
             provider="local",
             online_store=SqliteOnlineStoreConfig(path=online_store_path),
+            entity_key_serialization_version=2,
         )
     )
 
@@ -217,6 +218,7 @@ def feature_store_with_gcs_registry():
             registry=f"gs://{bucket_name}/registry.db",
             project="default",
             provider="gcp",
+            entity_key_serialization_version=2,
         )
     )
 
@@ -235,5 +237,6 @@ def feature_store_with_s3_registry():
                 region=os.getenv("AWS_REGION", "us-west-2")
             ),
             offline_store=FileOfflineStoreConfig(),
+            entity_key_serialization_version=2,
         )
     )