Add exception if materialize features defined on 'INPUT_CONTEXT' (#785)

* Add checking and error message if materialize features defined on INPUT_CONTEXT

feathr_project/feathr/client.py

@@ -32,6 +32,7 @@
 from feathr.utils._file_utils import write_to_file
 from feathr.utils.feature_printer import FeaturePrinter
 from feathr.utils.spark_job_params import FeatureGenerationJobParams, FeatureJoinJobParams
+from feathr.definition.source import InputContext
 
 
 class FeathrClient(object):
@@ -619,8 +620,15 @@ def materialize_features(self, settings: MaterializationSettings, execution_conf
             allow_materialize_non_agg_feature: Materializing non-aggregated features (the features without WindowAggTransformation) doesn't output meaningful results so it's by default set to False, but if you really want to materialize non-aggregated features, set this to True.
         """
         feature_list = settings.feature_names
-        if len(feature_list) > 0 and not self._valid_materialize_keys(feature_list):
-            raise RuntimeError(f"Invalid materialization features: {feature_list}, since they have different keys. Currently Feathr only supports materializing features of the same keys.")
+        if len(feature_list) > 0:
+            if 'anchor_list' in dir(self):
+                anchors = [anchor for anchor in self.anchor_list if isinstance(anchor.source, InputContext)]
+                anchor_feature_names = set(feature.name  for anchor in anchors for feature in anchor.features)
+                for feature in feature_list:
+                    if feature in anchor_feature_names:
+                        raise RuntimeError(f"Materializing features that are defined on INPUT_CONTEXT is not supported. {feature} is defined on INPUT_CONTEXT so you should remove it from the feature list in MaterializationSettings.")
+            if not self._valid_materialize_keys(feature_list):
+                raise RuntimeError(f"Invalid materialization features: {feature_list}, since they have different keys. Currently Feathr only supports materializing features of the same keys.")
 
         if not allow_materialize_non_agg_feature:
             # Check if there are non-aggregation features in the list

feathr_project/test/test_feature_materialization.py

@@ -18,6 +18,8 @@
 from test_fixture import basic_test_setup
 from test_fixture import get_online_test_table_name
 from test_utils.constants import Constants
+from logging import raiseExceptions
+import pytest
 
 def test_feature_materialization_config():
     backfill_time = BackfillTime(start=datetime(2020, 5, 20), end=datetime(2020, 5,20), step=timedelta(days=1))
@@ -255,4 +257,21 @@ def test_delete_feature_from_redis():
     res = client.get_online_features(online_test_table, '265', ['f_location_avg_fare'])
 
     assert len(res) == 1
-    assert res[0] == None
+    assert res[0] == None
+
+def test_feature_list_on_input_context():
+    with pytest.raises(RuntimeError) as e_info:
+        test_workspace_dir = Path(__file__).parent.resolve() / "test_user_workspace"
+
+        client: FeathrClient = basic_test_setup(os.path.join(test_workspace_dir, "feathr_config.yaml"))
+        online_test_table = get_online_test_table_name('nycTaxiCITableDeletion')
+        redisSink = RedisSink(table_name=online_test_table)
+        settings = MaterializationSettings(name="py_udf",
+                                       sinks=[redisSink],
+                                       feature_names=[
+                                           "f_location_avg_fare",
+                                           "f_day_of_week"
+                                       ])
+        client.materialize_features(settings, allow_materialize_non_agg_feature=True)
+    assert e_info is not None
+    assert e_info.value.args[0] == "Materializing features that are defined on INPUT_CONTEXT is not supported. f_day_of_week is defined on INPUT_CONTEXT so you should remove it from the feature list in MaterializationSettings."