We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Snowflake materialization fails on feature views with dummy entity
## Expected Behavior 

Same as feature views with entities

## Current Behavior

Materialization query errors, query has duplicate column names caused by a dummy entity having multiple dummy join keys

## Steps to reproduce

### Specifications

- Version:
- Platform:
- Subsystem:

## Possible Solution

even if infer features is ran more than once, only update the dummy entity join key once


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `sdk/python/feast/inference.py`
Content:
```
1 import re
2 from typing import List, Set, Union
3 
4 from feast.data_source import DataSource, PushSource, RequestSource
5 from feast.entity import Entity
6 from feast.errors import RegistryInferenceFailure
7 from feast.feature_view import DUMMY_ENTITY_ID, DUMMY_ENTITY_NAME, FeatureView
8 from feast.field import Field, from_value_type
9 from feast.infra.offline_stores.bigquery_source import BigQuerySource
10 from feast.infra.offline_stores.contrib.mssql_offline_store.mssqlserver_source import (
11     MsSqlServerSource,
12 )
13 from feast.infra.offline_stores.file_source import FileSource
14 from feast.infra.offline_stores.redshift_source import RedshiftSource
15 from feast.infra.offline_stores.snowflake_source import SnowflakeSource
16 from feast.repo_config import RepoConfig
17 from feast.stream_feature_view import StreamFeatureView
18 from feast.types import String
19 from feast.value_type import ValueType
20 
21 
22 def update_data_sources_with_inferred_event_timestamp_col(
23     data_sources: List[DataSource], config: RepoConfig
24 ) -> None:
25     ERROR_MSG_PREFIX = "Unable to infer DataSource timestamp_field"
26     for data_source in data_sources:
27         if isinstance(data_source, RequestSource):
28             continue
29         if isinstance(data_source, PushSource):
30             data_source = data_source.batch_source
31         if data_source.timestamp_field is None or data_source.timestamp_field == "":
32             # prepare right match pattern for data source
33             ts_column_type_regex_pattern: str
34             # TODO(adchia): Move Spark source inference out of this logic
35             if (
36                 isinstance(data_source, FileSource)
37                 or "SparkSource" == data_source.__class__.__name__
38             ):
39                 ts_column_type_regex_pattern = r"^timestamp"
40             elif isinstance(data_source, BigQuerySource):
41                 ts_column_type_regex_pattern = "TIMESTAMP|DATETIME"
42             elif isinstance(data_source, RedshiftSource):
43                 ts_column_type_regex_pattern = "TIMESTAMP[A-Z]*"
44             elif isinstance(data_source, SnowflakeSource):
45                 ts_column_type_regex_pattern = "TIMESTAMP_[A-Z]*"
46             elif isinstance(data_source, MsSqlServerSource):
47                 ts_column_type_regex_pattern = "TIMESTAMP|DATETIME"
48             else:
49                 raise RegistryInferenceFailure(
50                     "DataSource",
51                     f"""
52                     DataSource inferencing of timestamp_field is currently only supported
53                     for FileSource, SparkSource, BigQuerySource, RedshiftSource, SnowflakeSource, MsSqlSource.
54                     Attempting to infer from {data_source}.
55                     """,
56                 )
57             #  for informing the type checker
58             assert (
59                 isinstance(data_source, FileSource)
60                 or isinstance(data_source, BigQuerySource)
61                 or isinstance(data_source, RedshiftSource)
62                 or isinstance(data_source, SnowflakeSource)
63                 or isinstance(data_source, MsSqlServerSource)
64                 or "SparkSource" == data_source.__class__.__name__
65             )
66 
67             # loop through table columns to find singular match
68             timestamp_fields = []
69             for (
70                 col_name,
71                 col_datatype,
72             ) in data_source.get_table_column_names_and_types(config):
73                 if re.match(ts_column_type_regex_pattern, col_datatype):
74                     timestamp_fields.append(col_name)
75 
76             if len(timestamp_fields) > 1:
77                 raise RegistryInferenceFailure(
78                     "DataSource",
79                     f"""{ERROR_MSG_PREFIX}; found multiple possible columns of timestamp type.
80                     Data source type: {data_source.__class__.__name__},
81                     Timestamp regex: `{ts_column_type_regex_pattern}`, columns: {timestamp_fields}""",
82                 )
83             elif len(timestamp_fields) == 1:
84                 data_source.timestamp_field = timestamp_fields[0]
85             else:
86                 raise RegistryInferenceFailure(
87                     "DataSource",
88                     f"""
89                     {ERROR_MSG_PREFIX}; Found no columns of timestamp type.
90                     Data source type: {data_source.__class__.__name__},
91                     Timestamp regex: `{ts_column_type_regex_pattern}`.
92                     """,
93                 )
94 
95 
96 def update_feature_views_with_inferred_features_and_entities(
97     fvs: Union[List[FeatureView], List[StreamFeatureView]],
98     entities: List[Entity],
99     config: RepoConfig,
100 ) -> None:
101     """
102     Infers the features and entities associated with each feature view and updates it in place.
103 
104     Columns whose names match a join key of an entity are considered to be entity columns; all
105     other columns except designated timestamp columns are considered to be feature columns. If
106     the feature view already has features, feature inference is skipped.
107 
108     Note that this inference logic currently does not take any transformations (either a UDF or
109     aggregations) into account. For example, even if a stream feature view has a transformation,
110     this method assumes that the batch source contains transformed data with the correct final schema.
111 
112     Args:
113         fvs: The feature views to be updated.
114         entities: A list containing entities associated with the feature views.
115         config: The config for the current feature store.
116     """
117     entity_name_to_entity_map = {e.name: e for e in entities}
118     entity_name_to_join_keys_map = {e.name: e.join_keys for e in entities}
119 
120     for fv in fvs:
121         join_keys = set(
122             [
123                 join_key
124                 for entity_name in fv.entities
125                 for join_key in entity_name_to_join_keys_map[entity_name]
126             ]
127         )
128 
129         # Fields whose names match a join key are considered to be entity columns; all
130         # other fields are considered to be feature columns.
131         for field in fv.schema:
132             if field.name in join_keys:
133                 # Do not override a preexisting field with the same name.
134                 if field.name not in [
135                     entity_column.name for entity_column in fv.entity_columns
136                 ]:
137                     fv.entity_columns.append(field)
138             else:
139                 if field.name not in [feature.name for feature in fv.features]:
140                     fv.features.append(field)
141 
142         # Since the `value_type` parameter has not yet been fully deprecated for
143         # entities, we respect the `value_type` attribute if it still exists.
144         for entity_name in fv.entities:
145             entity = entity_name_to_entity_map[entity_name]
146             if (
147                 entity.join_key
148                 not in [entity_column.name for entity_column in fv.entity_columns]
149                 and entity.value_type != ValueType.UNKNOWN
150             ):
151                 fv.entity_columns.append(
152                     Field(
153                         name=entity.join_key,
154                         dtype=from_value_type(entity.value_type),
155                     )
156                 )
157 
158         # Infer a dummy entity column for entityless feature views.
159         if len(fv.entities) == 1 and fv.entities[0] == DUMMY_ENTITY_NAME:
160             fv.entity_columns.append(Field(name=DUMMY_ENTITY_ID, dtype=String))
161 
162         # Run inference for entity columns if there are fewer entity fields than expected.
163         num_expected_join_keys = sum(
164             [
165                 len(entity_name_to_join_keys_map[entity_name])
166                 for entity_name in fv.entities
167             ]
168         )
169         run_inference_for_entities = len(fv.entity_columns) < num_expected_join_keys
170 
171         # Run inference for feature columns if there are no feature fields.
172         run_inference_for_features = len(fv.features) == 0
173 
174         if run_inference_for_entities or run_inference_for_features:
175             _infer_features_and_entities(
176                 fv,
177                 join_keys,
178                 run_inference_for_features,
179                 config,
180             )
181 
182             if not fv.features:
183                 raise RegistryInferenceFailure(
184                     "FeatureView",
185                     f"Could not infer Features for the FeatureView named {fv.name}.",
186                 )
187 
188 
189 def _infer_features_and_entities(
190     fv: FeatureView,
191     join_keys: Set[str],
192     run_inference_for_features,
193     config,
194 ) -> None:
195     """
196     Updates the specific feature in place with inferred features and entities.
197 
198     Args:
199         fv: The feature view on which to run inference.
200         join_keys: The set of join keys for the feature view's entities.
201         run_inference_for_features: Whether to run inference for features.
202         config: The config for the current feature store.
203     """
204     columns_to_exclude = {
205         fv.batch_source.timestamp_field,
206         fv.batch_source.created_timestamp_column,
207     }
208     for original_col, mapped_col in fv.batch_source.field_mapping.items():
209         if mapped_col in columns_to_exclude:
210             columns_to_exclude.remove(mapped_col)
211             columns_to_exclude.add(original_col)
212 
213     table_column_names_and_types = fv.batch_source.get_table_column_names_and_types(
214         config
215     )
216 
217     for col_name, col_datatype in table_column_names_and_types:
218         if col_name in columns_to_exclude:
219             continue
220         elif col_name in join_keys:
221             field = Field(
222                 name=col_name,
223                 dtype=from_value_type(
224                     fv.batch_source.source_datatype_to_feast_value_type()(col_datatype)
225                 ),
226             )
227             if field.name not in [
228                 entity_column.name for entity_column in fv.entity_columns
229             ]:
230                 fv.entity_columns.append(field)
231         elif not re.match(
232             "^__|__$", col_name
233         ):  # double underscores often signal an internal-use column
234             if run_inference_for_features:
235                 feature_name = (
236                     fv.batch_source.field_mapping[col_name]
237                     if col_name in fv.batch_source.field_mapping
238                     else col_name
239                 )
240                 field = Field(
241                     name=feature_name,
242                     dtype=from_value_type(
243                         fv.batch_source.source_datatype_to_feast_value_type()(
244                             col_datatype
245                         )
246                     ),
247                 )
248                 if field.name not in [feature.name for feature in fv.features]:
249                     fv.features.append(field)
250
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/sdk/python/feast/inference.py b/sdk/python/feast/inference.py
--- a/sdk/python/feast/inference.py
+++ b/sdk/python/feast/inference.py
@@ -156,7 +156,11 @@
                 )
 
         # Infer a dummy entity column for entityless feature views.
-        if len(fv.entities) == 1 and fv.entities[0] == DUMMY_ENTITY_NAME:
+        if (
+            len(fv.entities) == 1
+            and fv.entities[0] == DUMMY_ENTITY_NAME
+            and not fv.entity_columns
+        ):
             fv.entity_columns.append(Field(name=DUMMY_ENTITY_ID, dtype=String))
 
         # Run inference for entity columns if there are fewer entity fields than expected.

{"golden_diff": "diff --git a/sdk/python/feast/inference.py b/sdk/python/feast/inference.py\n--- a/sdk/python/feast/inference.py\n+++ b/sdk/python/feast/inference.py\n@@ -156,7 +156,11 @@\n                 )\n \n         # Infer a dummy entity column for entityless feature views.\n-        if len(fv.entities) == 1 and fv.entities[0] == DUMMY_ENTITY_NAME:\n+        if (\n+            len(fv.entities) == 1\n+            and fv.entities[0] == DUMMY_ENTITY_NAME\n+            and not fv.entity_columns\n+        ):\n             fv.entity_columns.append(Field(name=DUMMY_ENTITY_ID, dtype=String))\n \n         # Run inference for entity columns if there are fewer entity fields than expected.\n", "issue": "Snowflake materialization fails on feature views with dummy entity\n## Expected Behavior \r\n\r\nSame as feature views with entities\r\n\r\n## Current Behavior\r\n\r\nMaterialization query errors, query has duplicate column names caused by a dummy entity having multiple dummy join keys\r\n\r\n## Steps to reproduce\r\n\r\n### Specifications\r\n\r\n- Version:\r\n- Platform:\r\n- Subsystem:\r\n\r\n## Possible Solution\r\n\r\neven if infer features is ran more than once, only update the dummy entity join key once\r\n\n", "before_files": [{"content": "import re\nfrom typing import List, Set, Union\n\nfrom feast.data_source import DataSource, PushSource, RequestSource\nfrom feast.entity import Entity\nfrom feast.errors import RegistryInferenceFailure\nfrom feast.feature_view import DUMMY_ENTITY_ID, DUMMY_ENTITY_NAME, FeatureView\nfrom feast.field import Field, from_value_type\nfrom feast.infra.offline_stores.bigquery_source import BigQuerySource\nfrom feast.infra.offline_stores.contrib.mssql_offline_store.mssqlserver_source import (\n    MsSqlServerSource,\n)\nfrom feast.infra.offline_stores.file_source import FileSource\nfrom feast.infra.offline_stores.redshift_source import RedshiftSource\nfrom feast.infra.offline_stores.snowflake_source import SnowflakeSource\nfrom feast.repo_config import RepoConfig\nfrom feast.stream_feature_view import StreamFeatureView\nfrom feast.types import String\nfrom feast.value_type import ValueType\n\n\ndef update_data_sources_with_inferred_event_timestamp_col(\n    data_sources: List[DataSource], config: RepoConfig\n) -> None:\n    ERROR_MSG_PREFIX = \"Unable to infer DataSource timestamp_field\"\n    for data_source in data_sources:\n        if isinstance(data_source, RequestSource):\n            continue\n        if isinstance(data_source, PushSource):\n            data_source = data_source.batch_source\n        if data_source.timestamp_field is None or data_source.timestamp_field == \"\":\n            # prepare right match pattern for data source\n            ts_column_type_regex_pattern: str\n            # TODO(adchia): Move Spark source inference out of this logic\n            if (\n                isinstance(data_source, FileSource)\n                or \"SparkSource\" == data_source.__class__.__name__\n            ):\n                ts_column_type_regex_pattern = r\"^timestamp\"\n            elif isinstance(data_source, BigQuerySource):\n                ts_column_type_regex_pattern = \"TIMESTAMP|DATETIME\"\n            elif isinstance(data_source, RedshiftSource):\n                ts_column_type_regex_pattern = \"TIMESTAMP[A-Z]*\"\n            elif isinstance(data_source, SnowflakeSource):\n                ts_column_type_regex_pattern = \"TIMESTAMP_[A-Z]*\"\n            elif isinstance(data_source, MsSqlServerSource):\n                ts_column_type_regex_pattern = \"TIMESTAMP|DATETIME\"\n            else:\n                raise RegistryInferenceFailure(\n                    \"DataSource\",\n                    f\"\"\"\n                    DataSource inferencing of timestamp_field is currently only supported\n                    for FileSource, SparkSource, BigQuerySource, RedshiftSource, SnowflakeSource, MsSqlSource.\n                    Attempting to infer from {data_source}.\n                    \"\"\",\n                )\n            #  for informing the type checker\n            assert (\n                isinstance(data_source, FileSource)\n                or isinstance(data_source, BigQuerySource)\n                or isinstance(data_source, RedshiftSource)\n                or isinstance(data_source, SnowflakeSource)\n                or isinstance(data_source, MsSqlServerSource)\n                or \"SparkSource\" == data_source.__class__.__name__\n            )\n\n            # loop through table columns to find singular match\n            timestamp_fields = []\n            for (\n                col_name,\n                col_datatype,\n            ) in data_source.get_table_column_names_and_types(config):\n                if re.match(ts_column_type_regex_pattern, col_datatype):\n                    timestamp_fields.append(col_name)\n\n            if len(timestamp_fields) > 1:\n                raise RegistryInferenceFailure(\n                    \"DataSource\",\n                    f\"\"\"{ERROR_MSG_PREFIX}; found multiple possible columns of timestamp type.\n                    Data source type: {data_source.__class__.__name__},\n                    Timestamp regex: `{ts_column_type_regex_pattern}`, columns: {timestamp_fields}\"\"\",\n                )\n            elif len(timestamp_fields) == 1:\n                data_source.timestamp_field = timestamp_fields[0]\n            else:\n                raise RegistryInferenceFailure(\n                    \"DataSource\",\n                    f\"\"\"\n                    {ERROR_MSG_PREFIX}; Found no columns of timestamp type.\n                    Data source type: {data_source.__class__.__name__},\n                    Timestamp regex: `{ts_column_type_regex_pattern}`.\n                    \"\"\",\n                )\n\n\ndef update_feature_views_with_inferred_features_and_entities(\n    fvs: Union[List[FeatureView], List[StreamFeatureView]],\n    entities: List[Entity],\n    config: RepoConfig,\n) -> None:\n    \"\"\"\n    Infers the features and entities associated with each feature view and updates it in place.\n\n    Columns whose names match a join key of an entity are considered to be entity columns; all\n    other columns except designated timestamp columns are considered to be feature columns. If\n    the feature view already has features, feature inference is skipped.\n\n    Note that this inference logic currently does not take any transformations (either a UDF or\n    aggregations) into account. For example, even if a stream feature view has a transformation,\n    this method assumes that the batch source contains transformed data with the correct final schema.\n\n    Args:\n        fvs: The feature views to be updated.\n        entities: A list containing entities associated with the feature views.\n        config: The config for the current feature store.\n    \"\"\"\n    entity_name_to_entity_map = {e.name: e for e in entities}\n    entity_name_to_join_keys_map = {e.name: e.join_keys for e in entities}\n\n    for fv in fvs:\n        join_keys = set(\n            [\n                join_key\n                for entity_name in fv.entities\n                for join_key in entity_name_to_join_keys_map[entity_name]\n            ]\n        )\n\n        # Fields whose names match a join key are considered to be entity columns; all\n        # other fields are considered to be feature columns.\n        for field in fv.schema:\n            if field.name in join_keys:\n                # Do not override a preexisting field with the same name.\n                if field.name not in [\n                    entity_column.name for entity_column in fv.entity_columns\n                ]:\n                    fv.entity_columns.append(field)\n            else:\n                if field.name not in [feature.name for feature in fv.features]:\n                    fv.features.append(field)\n\n        # Since the `value_type` parameter has not yet been fully deprecated for\n        # entities, we respect the `value_type` attribute if it still exists.\n        for entity_name in fv.entities:\n            entity = entity_name_to_entity_map[entity_name]\n            if (\n                entity.join_key\n                not in [entity_column.name for entity_column in fv.entity_columns]\n                and entity.value_type != ValueType.UNKNOWN\n            ):\n                fv.entity_columns.append(\n                    Field(\n                        name=entity.join_key,\n                        dtype=from_value_type(entity.value_type),\n                    )\n                )\n\n        # Infer a dummy entity column for entityless feature views.\n        if len(fv.entities) == 1 and fv.entities[0] == DUMMY_ENTITY_NAME:\n            fv.entity_columns.append(Field(name=DUMMY_ENTITY_ID, dtype=String))\n\n        # Run inference for entity columns if there are fewer entity fields than expected.\n        num_expected_join_keys = sum(\n            [\n                len(entity_name_to_join_keys_map[entity_name])\n                for entity_name in fv.entities\n            ]\n        )\n        run_inference_for_entities = len(fv.entity_columns) < num_expected_join_keys\n\n        # Run inference for feature columns if there are no feature fields.\n        run_inference_for_features = len(fv.features) == 0\n\n        if run_inference_for_entities or run_inference_for_features:\n            _infer_features_and_entities(\n                fv,\n                join_keys,\n                run_inference_for_features,\n                config,\n            )\n\n            if not fv.features:\n                raise RegistryInferenceFailure(\n                    \"FeatureView\",\n                    f\"Could not infer Features for the FeatureView named {fv.name}.\",\n                )\n\n\ndef _infer_features_and_entities(\n    fv: FeatureView,\n    join_keys: Set[str],\n    run_inference_for_features,\n    config,\n) -> None:\n    \"\"\"\n    Updates the specific feature in place with inferred features and entities.\n\n    Args:\n        fv: The feature view on which to run inference.\n        join_keys: The set of join keys for the feature view's entities.\n        run_inference_for_features: Whether to run inference for features.\n        config: The config for the current feature store.\n    \"\"\"\n    columns_to_exclude = {\n        fv.batch_source.timestamp_field,\n        fv.batch_source.created_timestamp_column,\n    }\n    for original_col, mapped_col in fv.batch_source.field_mapping.items():\n        if mapped_col in columns_to_exclude:\n            columns_to_exclude.remove(mapped_col)\n            columns_to_exclude.add(original_col)\n\n    table_column_names_and_types = fv.batch_source.get_table_column_names_and_types(\n        config\n    )\n\n    for col_name, col_datatype in table_column_names_and_types:\n        if col_name in columns_to_exclude:\n            continue\n        elif col_name in join_keys:\n            field = Field(\n                name=col_name,\n                dtype=from_value_type(\n                    fv.batch_source.source_datatype_to_feast_value_type()(col_datatype)\n                ),\n            )\n            if field.name not in [\n                entity_column.name for entity_column in fv.entity_columns\n            ]:\n                fv.entity_columns.append(field)\n        elif not re.match(\n            \"^__|__$\", col_name\n        ):  # double underscores often signal an internal-use column\n            if run_inference_for_features:\n                feature_name = (\n                    fv.batch_source.field_mapping[col_name]\n                    if col_name in fv.batch_source.field_mapping\n                    else col_name\n                )\n                field = Field(\n                    name=feature_name,\n                    dtype=from_value_type(\n                        fv.batch_source.source_datatype_to_feast_value_type()(\n                            col_datatype\n                        )\n                    ),\n                )\n                if field.name not in [feature.name for feature in fv.features]:\n                    fv.features.append(field)\n", "path": "sdk/python/feast/inference.py"}], "after_files": [{"content": "import re\nfrom typing import List, Set, Union\n\nfrom feast.data_source import DataSource, PushSource, RequestSource\nfrom feast.entity import Entity\nfrom feast.errors import RegistryInferenceFailure\nfrom feast.feature_view import DUMMY_ENTITY_ID, DUMMY_ENTITY_NAME, FeatureView\nfrom feast.field import Field, from_value_type\nfrom feast.infra.offline_stores.bigquery_source import BigQuerySource\nfrom feast.infra.offline_stores.contrib.mssql_offline_store.mssqlserver_source import (\n    MsSqlServerSource,\n)\nfrom feast.infra.offline_stores.file_source import FileSource\nfrom feast.infra.offline_stores.redshift_source import RedshiftSource\nfrom feast.infra.offline_stores.snowflake_source import SnowflakeSource\nfrom feast.repo_config import RepoConfig\nfrom feast.stream_feature_view import StreamFeatureView\nfrom feast.types import String\nfrom feast.value_type import ValueType\n\n\ndef update_data_sources_with_inferred_event_timestamp_col(\n    data_sources: List[DataSource], config: RepoConfig\n) -> None:\n    ERROR_MSG_PREFIX = \"Unable to infer DataSource timestamp_field\"\n    for data_source in data_sources:\n        if isinstance(data_source, RequestSource):\n            continue\n        if isinstance(data_source, PushSource):\n            data_source = data_source.batch_source\n        if data_source.timestamp_field is None or data_source.timestamp_field == \"\":\n            # prepare right match pattern for data source\n            ts_column_type_regex_pattern: str\n            # TODO(adchia): Move Spark source inference out of this logic\n            if (\n                isinstance(data_source, FileSource)\n                or \"SparkSource\" == data_source.__class__.__name__\n            ):\n                ts_column_type_regex_pattern = r\"^timestamp\"\n            elif isinstance(data_source, BigQuerySource):\n                ts_column_type_regex_pattern = \"TIMESTAMP|DATETIME\"\n            elif isinstance(data_source, RedshiftSource):\n                ts_column_type_regex_pattern = \"TIMESTAMP[A-Z]*\"\n            elif isinstance(data_source, SnowflakeSource):\n                ts_column_type_regex_pattern = \"TIMESTAMP_[A-Z]*\"\n            elif isinstance(data_source, MsSqlServerSource):\n                ts_column_type_regex_pattern = \"TIMESTAMP|DATETIME\"\n            else:\n                raise RegistryInferenceFailure(\n                    \"DataSource\",\n                    f\"\"\"\n                    DataSource inferencing of timestamp_field is currently only supported\n                    for FileSource, SparkSource, BigQuerySource, RedshiftSource, SnowflakeSource, MsSqlSource.\n                    Attempting to infer from {data_source}.\n                    \"\"\",\n                )\n            #  for informing the type checker\n            assert (\n                isinstance(data_source, FileSource)\n                or isinstance(data_source, BigQuerySource)\n                or isinstance(data_source, RedshiftSource)\n                or isinstance(data_source, SnowflakeSource)\n                or isinstance(data_source, MsSqlServerSource)\n                or \"SparkSource\" == data_source.__class__.__name__\n            )\n\n            # loop through table columns to find singular match\n            timestamp_fields = []\n            for (\n                col_name,\n                col_datatype,\n            ) in data_source.get_table_column_names_and_types(config):\n                if re.match(ts_column_type_regex_pattern, col_datatype):\n                    timestamp_fields.append(col_name)\n\n            if len(timestamp_fields) > 1:\n                raise RegistryInferenceFailure(\n                    \"DataSource\",\n                    f\"\"\"{ERROR_MSG_PREFIX}; found multiple possible columns of timestamp type.\n                    Data source type: {data_source.__class__.__name__},\n                    Timestamp regex: `{ts_column_type_regex_pattern}`, columns: {timestamp_fields}\"\"\",\n                )\n            elif len(timestamp_fields) == 1:\n                data_source.timestamp_field = timestamp_fields[0]\n            else:\n                raise RegistryInferenceFailure(\n                    \"DataSource\",\n                    f\"\"\"\n                    {ERROR_MSG_PREFIX}; Found no columns of timestamp type.\n                    Data source type: {data_source.__class__.__name__},\n                    Timestamp regex: `{ts_column_type_regex_pattern}`.\n                    \"\"\",\n                )\n\n\ndef update_feature_views_with_inferred_features_and_entities(\n    fvs: Union[List[FeatureView], List[StreamFeatureView]],\n    entities: List[Entity],\n    config: RepoConfig,\n) -> None:\n    \"\"\"\n    Infers the features and entities associated with each feature view and updates it in place.\n\n    Columns whose names match a join key of an entity are considered to be entity columns; all\n    other columns except designated timestamp columns are considered to be feature columns. If\n    the feature view already has features, feature inference is skipped.\n\n    Note that this inference logic currently does not take any transformations (either a UDF or\n    aggregations) into account. For example, even if a stream feature view has a transformation,\n    this method assumes that the batch source contains transformed data with the correct final schema.\n\n    Args:\n        fvs: The feature views to be updated.\n        entities: A list containing entities associated with the feature views.\n        config: The config for the current feature store.\n    \"\"\"\n    entity_name_to_entity_map = {e.name: e for e in entities}\n    entity_name_to_join_keys_map = {e.name: e.join_keys for e in entities}\n\n    for fv in fvs:\n        join_keys = set(\n            [\n                join_key\n                for entity_name in fv.entities\n                for join_key in entity_name_to_join_keys_map[entity_name]\n            ]\n        )\n\n        # Fields whose names match a join key are considered to be entity columns; all\n        # other fields are considered to be feature columns.\n        for field in fv.schema:\n            if field.name in join_keys:\n                # Do not override a preexisting field with the same name.\n                if field.name not in [\n                    entity_column.name for entity_column in fv.entity_columns\n                ]:\n                    fv.entity_columns.append(field)\n            else:\n                if field.name not in [feature.name for feature in fv.features]:\n                    fv.features.append(field)\n\n        # Since the `value_type` parameter has not yet been fully deprecated for\n        # entities, we respect the `value_type` attribute if it still exists.\n        for entity_name in fv.entities:\n            entity = entity_name_to_entity_map[entity_name]\n            if (\n                entity.join_key\n                not in [entity_column.name for entity_column in fv.entity_columns]\n                and entity.value_type != ValueType.UNKNOWN\n            ):\n                fv.entity_columns.append(\n                    Field(\n                        name=entity.join_key,\n                        dtype=from_value_type(entity.value_type),\n                    )\n                )\n\n        # Infer a dummy entity column for entityless feature views.\n        if (\n            len(fv.entities) == 1\n            and fv.entities[0] == DUMMY_ENTITY_NAME\n            and not fv.entity_columns\n        ):\n            fv.entity_columns.append(Field(name=DUMMY_ENTITY_ID, dtype=String))\n\n        # Run inference for entity columns if there are fewer entity fields than expected.\n        num_expected_join_keys = sum(\n            [\n                len(entity_name_to_join_keys_map[entity_name])\n                for entity_name in fv.entities\n            ]\n        )\n        run_inference_for_entities = len(fv.entity_columns) < num_expected_join_keys\n\n        # Run inference for feature columns if there are no feature fields.\n        run_inference_for_features = len(fv.features) == 0\n\n        if run_inference_for_entities or run_inference_for_features:\n            _infer_features_and_entities(\n                fv,\n                join_keys,\n                run_inference_for_features,\n                config,\n            )\n\n            if not fv.features:\n                raise RegistryInferenceFailure(\n                    \"FeatureView\",\n                    f\"Could not infer Features for the FeatureView named {fv.name}.\",\n                )\n\n\ndef _infer_features_and_entities(\n    fv: FeatureView,\n    join_keys: Set[str],\n    run_inference_for_features,\n    config,\n) -> None:\n    \"\"\"\n    Updates the specific feature in place with inferred features and entities.\n\n    Args:\n        fv: The feature view on which to run inference.\n        join_keys: The set of join keys for the feature view's entities.\n        run_inference_for_features: Whether to run inference for features.\n        config: The config for the current feature store.\n    \"\"\"\n    columns_to_exclude = {\n        fv.batch_source.timestamp_field,\n        fv.batch_source.created_timestamp_column,\n    }\n    for original_col, mapped_col in fv.batch_source.field_mapping.items():\n        if mapped_col in columns_to_exclude:\n            columns_to_exclude.remove(mapped_col)\n            columns_to_exclude.add(original_col)\n\n    table_column_names_and_types = fv.batch_source.get_table_column_names_and_types(\n        config\n    )\n\n    for col_name, col_datatype in table_column_names_and_types:\n        if col_name in columns_to_exclude:\n            continue\n        elif col_name in join_keys:\n            field = Field(\n                name=col_name,\n                dtype=from_value_type(\n                    fv.batch_source.source_datatype_to_feast_value_type()(col_datatype)\n                ),\n            )\n            if field.name not in [\n                entity_column.name for entity_column in fv.entity_columns\n            ]:\n                fv.entity_columns.append(field)\n        elif not re.match(\n            \"^__|__$\", col_name\n        ):  # double underscores often signal an internal-use column\n            if run_inference_for_features:\n                feature_name = (\n                    fv.batch_source.field_mapping[col_name]\n                    if col_name in fv.batch_source.field_mapping\n                    else col_name\n                )\n                field = Field(\n                    name=feature_name,\n                    dtype=from_value_type(\n                        fv.batch_source.source_datatype_to_feast_value_type()(\n                            col_datatype\n                        )\n                    ),\n                )\n                if field.name not in [feature.name for feature in fv.features]:\n                    fv.features.append(field)\n", "path": "sdk/python/feast/inference.py"}]}