Visitar URL original
feat: Updating FeatureViewProjection and OnDemandFeatureView to add batch_source and entities by franciscojavierarceo · Pull Request #4530 · feast-dev/feast · GitHub
Skip to content

feat: Updating FeatureViewProjection and OnDemandFeatureView to add batch_source and entities #4530

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
haoxu0 merged 16 commits into from
Sep 23, 2024
Merged

feat: Updating FeatureViewProjection and OnDemandFeatureView to add batch_source and entities #4530

Changes from 1 commit
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
b791284
feat: Updating protos for Projections to include more info
franciscojavierarceo Sep 17, 2024
54ca376
adding unit test
franciscojavierarceo Sep 18, 2024
cf613bd
adding type checking where batch source is already serialized into pr…
franciscojavierarceo Sep 18, 2024
44066d2
almost got everything working and type validation behaving
franciscojavierarceo Sep 18, 2024
2837c1c
cleaned up and have tests behaving
franciscojavierarceo Sep 18, 2024
52d4253
removed comment
franciscojavierarceo Sep 18, 2024
34f99b9
updated FeatureViewProjection batch_source serialization
franciscojavierarceo Sep 19, 2024
fcf2917
trying to debug a test
franciscojavierarceo Sep 19, 2024
3c7812b
handling snowflake issue, cant confirm why it is happening so just go…
franciscojavierarceo Sep 19, 2024
6c9a21f
linter
franciscojavierarceo Sep 19, 2024
1103af7
trying to handle it correctly
franciscojavierarceo Sep 19, 2024
987c690
handling the else case for from_feature_view_definition
franciscojavierarceo Sep 19, 2024
c5a7ea3
adding print
franciscojavierarceo Sep 21, 2024
0d489a9
adding test of issue
franciscojavierarceo Sep 21, 2024
a85854e
think i got everything working now
franciscojavierarceo Sep 21, 2024
cf86862
removing print
franciscojavierarceo Sep 21, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
think i got everything working now 
Signed-off-by: Francisco Javier Arceo <farceo@redhat.com>
  • Loading branch information
@franciscojavierarceo
franciscojavierarceo committed Sep 21, 2024
commit a85854e23d75bf253c6f468bbc632efd3300410b
173 changes: 91 additions & 82 deletions sdk/python/feast/inference.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -204,122 +204,131 @@ def _infer_features_and_entities(
run_inference_for_features: Whether to run inference for features.
config: The config for the current feature store.
"""
entity_columns: list[Field] = []
if isinstance(fv, OnDemandFeatureView):
columns_to_exclude = set()
for (
source_feature_view_name,
source_feature_view,
) in fv.source_feature_view_projections.items():
columns_to_exclude.add(source_feature_view.timestamp_field)
columns_to_exclude.add(source_feature_view.created_timestamp_column)
return _infer_on_demand_features_and_entities(
fv, join_keys, run_inference_for_features, config
)

batch_source = getattr(source_feature_view, "batch_source")
batch_field_mapping = getattr(batch_source or None, "field_mapping")
if batch_field_mapping:
for (
original_col,
mapped_col,
) in batch_field_mapping.items():
if mapped_col in columns_to_exclude:
columns_to_exclude.remove(mapped_col)
columns_to_exclude.add(original_col)
columns_to_exclude = {
fv.batch_source.timestamp_field,
fv.batch_source.created_timestamp_column,
}
for original_col, mapped_col in fv.batch_source.field_mapping.items():
if mapped_col in columns_to_exclude:
columns_to_exclude.remove(mapped_col)
columns_to_exclude.add(original_col)

table_column_names_and_types = (
batch_source.get_table_column_names_and_types(config)
)
table_column_names_and_types = fv.batch_source.get_table_column_names_and_types(
config
)

for col_name, col_datatype in table_column_names_and_types:
if col_name in columns_to_exclude:
continue
elif col_name in join_keys:
field = Field(
name=col_name,
dtype=from_value_type(
batch_source.source_datatype_to_feast_value_type()(
col_datatype
)
),
)
if field.name not in [
entity_column.name
for entity_column in entity_columns
if hasattr(entity_column, "name")
]:
entity_columns.append(field)
elif not re.match(
"^__|__$", col_name
): # double underscores often signal an internal-use column
if run_inference_for_features:
feature_name = (
batch_field_mapping[col_name]
if col_name in batch_field_mapping
else col_name
)
field = Field(
name=feature_name,
dtype=from_value_type(
batch_source.source_datatype_to_feast_value_type()(
col_datatype
)
),
for col_name, col_datatype in table_column_names_and_types:
if col_name in columns_to_exclude:
continue
elif col_name in join_keys:
field = Field(
name=col_name,
dtype=from_value_type(
fv.batch_source.source_datatype_to_feast_value_type()(col_datatype)
),
)
if field.name not in [
entity_column.name for entity_column in fv.entity_columns
]:
fv.entity_columns.append(field)
elif not re.match(
"^__|__$", col_name
): # double underscores often signal an internal-use column
if run_inference_for_features:
feature_name = (
fv.batch_source.field_mapping[col_name]
if col_name in fv.batch_source.field_mapping
else col_name
)
field = Field(
name=feature_name,
dtype=from_value_type(
fv.batch_source.source_datatype_to_feast_value_type()(
col_datatype
)
if field.name not in [
feature.name for feature in source_feature_view.features
]:
source_feature_view.features.append(field)
),
)
if field.name not in [feature.name for feature in fv.features]:
fv.features.append(field)

else:
columns_to_exclude = {
fv.batch_source.timestamp_field,
fv.batch_source.created_timestamp_column,
}
for original_col, mapped_col in fv.batch_source.field_mapping.items():
if mapped_col in columns_to_exclude:
columns_to_exclude.remove(mapped_col)
columns_to_exclude.add(original_col)

table_column_names_and_types = fv.batch_source.get_table_column_names_and_types(
config
)
def _infer_on_demand_features_and_entities(
fv: OnDemandFeatureView,
join_keys: Set[Optional[str]],
run_inference_for_features,
config,
) -> None:
"""
Updates the specific feature in place with inferred features and entities.
Args:
fv: The feature view on which to run inference.
join_keys: The set of join keys for the feature view's entities.
run_inference_for_features: Whether to run inference for features.
config: The config for the current feature store.
"""
entity_columns: list[Field] = []
columns_to_exclude = set()
for (
source_feature_view_name,
source_feature_view,
) in fv.source_feature_view_projections.items():
columns_to_exclude.add(source_feature_view.timestamp_field)
columns_to_exclude.add(source_feature_view.created_timestamp_column)

batch_source = getattr(source_feature_view, "batch_source")
batch_field_mapping = getattr(batch_source or None, "field_mapping")
if batch_field_mapping:
for (
original_col,
mapped_col,
) in batch_field_mapping.items():
if mapped_col in columns_to_exclude:
columns_to_exclude.remove(mapped_col)
columns_to_exclude.add(original_col)

table_column_names_and_types = (
batch_source.get_table_column_names_and_types(config)
)
for col_name, col_datatype in table_column_names_and_types:
if col_name in columns_to_exclude:
continue
elif col_name in join_keys:
field = Field(
name=col_name,
dtype=from_value_type(
fv.batch_source.source_datatype_to_feast_value_type()(
col_datatype
)
batch_source.source_datatype_to_feast_value_type()(col_datatype)
),
)
if field.name not in [
entity_column.name
if not isinstance(entity_column, str)
else entity_column
for entity_column in entity_columns
if hasattr(entity_column, "name")
]:
entity_columns.append(field)
elif not re.match(
"^__|__$", col_name
): # double underscores often signal an internal-use column
if run_inference_for_features:
feature_name = (
fv.batch_source.field_mapping[col_name]
if col_name in fv.batch_source.field_mapping
batch_field_mapping[col_name]
if col_name in batch_field_mapping
else col_name
)
field = Field(
name=feature_name,
dtype=from_value_type(
fv.batch_source.source_datatype_to_feast_value_type()(
batch_source.source_datatype_to_feast_value_type()(
col_datatype
)
),
)
if field.name not in [feature.name for feature in fv.features]:
fv.features.append(field)

if field.name not in [
feature.name for feature in source_feature_view.features
]:
source_feature_view.features.append(field)
fv.entity_columns = entity_columns