Skip to content
Merged
7 changes: 4 additions & 3 deletions sdk/python/feast/feature_store.py
Original file line number Diff line number Diff line change
Expand Up @@ -1296,9 +1296,10 @@ def _get_feature_views_to_materialize(
f"Enable it before materializing."
)
if hasattr(feature_view, "online") and not feature_view.online:
raise ValueError(
f"FeatureView {feature_view.name} is not configured to be served online."
)
if not getattr(feature_view, "offline", False):
raise ValueError(
f"FeatureView {feature_view.name} is not configured to be served online."
)
Comment thread
abhijeet-dhumal marked this conversation as resolved.
elif (
hasattr(feature_view, "write_to_online_store")
and not feature_view.write_to_online_store
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -299,7 +299,9 @@ def get_historical_features(
)

query_context = _apply_bfv_transformations(
spark_session, feature_views, query_context
spark_session=spark_session,
feature_views=feature_views,
query_contexts=query_context,
)

spark_query_context = [
Expand Down Expand Up @@ -1406,9 +1408,16 @@ def _apply_bfv_transformations(
query_contexts: List[offline_utils.FeatureViewQueryContext],
) -> List[offline_utils.FeatureViewQueryContext]:
"""
For BatchFeatureViews with a UDF, read the raw source into a Spark DataFrame,
invoke the transformation, register the result as a temp view, and replace the
table_subquery in the query context so the PIT join reads transformed data.
For BatchFeatureViews, update each query context in one of two ways:

1. Pre-computed path shortcut: if ``offline=True`` and
``batch_source.path`` is set, read the pre-materialized parquet
directly — avoids re-running the UDF on every training call.
2. UDF execution: if the BFV has a transformation, run it against
the raw source and register the result as a temp view.

Plain FeatureViews and BFVs with neither a path nor a UDF pass
through unchanged.
"""
from dataclasses import replace

Expand All @@ -1423,11 +1432,41 @@ def _apply_bfv_transformations(
updated_contexts = []
for ctx in query_contexts:
fv = fv_by_name.get(ctx.name)
if fv is None or not isinstance(fv, BatchFeatureView):
updated_contexts.append(ctx)
continue

# 1. Pre-computed path shortcut
if (
fv is not None
and isinstance(fv, BatchFeatureView)
and has_transformation(fv)
getattr(fv, "offline", False)
and isinstance(fv.batch_source, SparkSource)
and fv.batch_source.path
):
tmp_view = f"__feast_offline_{ctx.name}_{uuid.uuid4().hex[:8]}"
file_format = fv.batch_source.file_format or "parquet"
try:
df = spark_session.read.format(file_format).load(fv.batch_source.path)
df.createOrReplaceTempView(tmp_view)
updated_contexts.append(replace(ctx, table_subquery=tmp_view))
continue
except (FileNotFoundError, PermissionError) as e:
warnings.warn(
f"Offline path '{fv.batch_source.path}' not accessible for "
f"'{ctx.name}': {e}; falling back to source query.",
RuntimeWarning,
stacklevel=2,
)
except Exception as e:
warnings.warn(
f"Unexpected error loading offline path "
f"'{fv.batch_source.path}' for '{ctx.name}': {e}; "
f"falling back to source query.",
RuntimeWarning,
stacklevel=2,
)

# 2. UDF execution fallback
if has_transformation(fv):
udf = get_transformation_function(fv)
if udf is not None:
source_info = resolve_feature_view_source_with_fallback(fv)
Expand All @@ -1443,12 +1482,9 @@ def _apply_bfv_transformations(
source_df = spark_session.sql(
f"SELECT * FROM {source_query} WHERE {timestamp_filter}"
)

transformed_df = udf(source_df)

tmp_view_name = "feast_bfv_" + uuid.uuid4().hex
transformed_df.createOrReplaceTempView(tmp_view_name)

ctx = replace(ctx, table_subquery=tmp_view_name)

updated_contexts.append(ctx)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -287,11 +287,17 @@ def __init__(
date_partition_column_format: Optional[str] = "%Y-%m-%d",
table_format: Optional[TableFormat] = None,
):
# Check that only one of the ways to load a spark dataframe can be used. We have
# to treat empty string and null the same due to proto (de)serialization.
if sum([(not (not arg)) for arg in [table, query, path]]) != 1:
# query + path is allowed: query for reads during materialization,
# path for offline write-back (offline=True) and get_historical_features.
# table must be standalone (cannot combine with query or path).
has_table = bool(table)
has_query = bool(query)
has_path = bool(path)
if has_table and (has_query or has_path):
raise ValueError("'table' cannot be combined with 'query' or 'path'.")
if not (has_table or has_query or has_path):
raise ValueError(
"Exactly one of params(table, query, path) must be specified."
"At least one of params(table, query, path) must be specified."
)
if path:
# If table_format is specified, file_format is optional (table format determines the reader)
Expand Down
2 changes: 1 addition & 1 deletion sdk/python/tests/unit/test_feature_server_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -677,7 +677,7 @@ def test_faster_than_message_to_dict(self):
print(f"\nPerformance: fast={fast_time:.3f}s, standard={standard_time:.3f}s")
print(f"Speedup: {speedup:.2f}x")

assert speedup >= 1.5, f"Expected at least 1.5x speedup, got {speedup:.2f}x"
assert speedup >= 1.2, f"Expected at least 1.2x speedup, got {speedup:.2f}x"


class TestStatusNames:
Expand Down
Loading