Skip to content

Commit 76af065

Browse files
ntkatholentkathole
committed
fix: Fixed logic for source/derived feature views
Signed-off-by: ntkathole <nikhilkathole2683@gmail.com>
1 parent dede2cd commit 76af065

14 files changed

Lines changed: 1210 additions & 427 deletions

File tree

‎Makefile‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -322,7 +322,7 @@ test-python-universal-ray-offline: ## Run Python Ray offline store integration t
322322

323323
test-python-ray-compute-engine: ## Run Python Ray compute engine tests
324324
PYTHONPATH='.' \
325-
python -m pytest --integration \
325+
python -m pytest -v --integration \
326326
sdk/python/tests/integration/compute_engines/ray_compute/
327327

328328
test-python-universal-postgres-online: ## Run Python Postgres integration tests

‎sdk/python/feast/feature_view.py‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -510,7 +510,8 @@ def _from_proto_internal(
510510
if feature_view_proto.spec.ttl.ToNanoseconds() == 0
511511
else feature_view_proto.spec.ttl.ToTimedelta()
512512
),
513-
source=batch_source if batch_source else source_views,
513+
source=source_views if source_views else batch_source,
514+
sink_source=batch_source if source_views else None,
514515
)
515516
if stream_source:
516517
feature_view.stream_source = stream_source

‎sdk/python/feast/feature_view_utils.py‎

Lines changed: 229 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,229 @@
1+
"""
2+
Utility functions for feature view operations including source resolution.
3+
"""
4+
5+
import logging
6+
import typing
7+
from dataclasses import dataclass
8+
from typing import Callable, Optional
9+
10+
if typing.TYPE_CHECKING:
11+
from feast.data_source import DataSource
12+
from feast.feature_view import FeatureView
13+
from feast.repo_config import RepoConfig
14+
15+
logger = logging.getLogger(__name__)
16+
17+
18+
@dataclass
19+
class FeatureViewSourceInfo:
20+
"""Information about a feature view's data source resolution."""
21+
22+
data_source: "DataSource"
23+
source_type: str
24+
has_transformation: bool
25+
transformation_func: Optional[Callable] = None
26+
source_description: str = ""
27+
28+
29+
def has_transformation(feature_view: "FeatureView") -> bool:
30+
"""Check if a feature view has transformations (UDF or feature_transformation)."""
31+
return (
32+
getattr(feature_view, "udf", None) is not None
33+
or getattr(feature_view, "feature_transformation", None) is not None
34+
)
35+
36+
37+
def get_transformation_function(feature_view: "FeatureView") -> Optional[Callable]:
38+
"""Extract the transformation function from a feature view."""
39+
feature_transformation = getattr(feature_view, "feature_transformation", None)
40+
if feature_transformation:
41+
# Use feature_transformation if available (preferred)
42+
if hasattr(feature_transformation, "udf") and callable(
43+
feature_transformation.udf
44+
):
45+
return feature_transformation.udf
46+
47+
# Fallback to direct UDF
48+
udf = getattr(feature_view, "udf", None)
49+
if udf and callable(udf):
50+
return udf
51+
52+
return None
53+
54+
55+
def find_original_source_view(feature_view: "FeatureView") -> "FeatureView":
56+
"""
57+
Recursively find the original source feature view that has a batch_source.
58+
For derived feature views, this follows the source_views chain until it finds
59+
a feature view with an actual DataSource (batch_source).
60+
"""
61+
current_view = feature_view
62+
while hasattr(current_view, "source_views") and current_view.source_views:
63+
if not current_view.source_views:
64+
break
65+
current_view = current_view.source_views[0] # Assuming single source for now
66+
return current_view
67+
68+
69+
def check_sink_source_exists(data_source: "DataSource") -> bool:
70+
"""
71+
Check if a sink_source file actually exists.
72+
Args:
73+
data_source: The DataSource to check
74+
Returns:
75+
bool: True if the source exists, False otherwise
76+
"""
77+
try:
78+
import fsspec
79+
80+
# Get the source path
81+
if hasattr(data_source, "path"):
82+
source_path = data_source.path
83+
else:
84+
source_path = str(data_source)
85+
86+
fs, path_in_fs = fsspec.core.url_to_fs(source_path)
87+
return fs.exists(path_in_fs)
88+
except Exception as e:
89+
logger.warning(f"Failed to check if source exists: {e}")
90+
return False
91+
92+
93+
def resolve_feature_view_source(
94+
feature_view: "FeatureView",
95+
config: Optional["RepoConfig"] = None,
96+
is_materialization: bool = False,
97+
) -> FeatureViewSourceInfo:
98+
"""
99+
Resolve the appropriate data source for a feature view.
100+
101+
This handles the complex logic of determining whether to read from:
102+
1. sink_source (materialized data from parent views)
103+
2. batch_source (original data source)
104+
3. Recursive resolution for derived views
105+
106+
Args:
107+
feature_view: The feature view to resolve
108+
config: Repository configuration (optional)
109+
is_materialization: Whether this is during materialization (affects derived view handling)
110+
111+
Returns:
112+
FeatureViewSourceInfo: Information about the resolved source
113+
"""
114+
view_has_transformation = has_transformation(feature_view)
115+
transformation_func = (
116+
get_transformation_function(feature_view) if view_has_transformation else None
117+
)
118+
119+
# Check if this is a derived feature view (has source_views)
120+
is_derived_view = (
121+
hasattr(feature_view, "source_views") and feature_view.source_views
122+
)
123+
124+
if not is_derived_view:
125+
# Regular feature view - use its batch_source directly
126+
return FeatureViewSourceInfo(
127+
data_source=feature_view.batch_source,
128+
source_type="batch_source",
129+
has_transformation=view_has_transformation,
130+
transformation_func=transformation_func,
131+
source_description=f"Direct batch_source for {feature_view.name}",
132+
)
133+
134+
# This is a derived feature view - need to resolve parent source
135+
if not feature_view.source_views:
136+
raise ValueError(
137+
f"Derived feature view {feature_view.name} has no source_views"
138+
)
139+
parent_view = feature_view.source_views[0] # Assuming single source for now
140+
141+
# For derived views: distinguish between materialization and historical retrieval
142+
if (
143+
hasattr(parent_view, "sink_source")
144+
and parent_view.sink_source
145+
and is_materialization
146+
):
147+
# During materialization, try to use sink_source if it exists
148+
if check_sink_source_exists(parent_view.sink_source):
149+
logger.debug(
150+
f"Materialization: Using parent {parent_view.name} sink_source"
151+
)
152+
return FeatureViewSourceInfo(
153+
data_source=parent_view.sink_source,
154+
source_type="sink_source",
155+
has_transformation=view_has_transformation,
156+
transformation_func=transformation_func,
157+
source_description=f"Parent {parent_view.name} sink_source for derived view {feature_view.name}",
158+
)
159+
else:
160+
logger.info(
161+
f"Parent {parent_view.name} sink_source doesn't exist during materialization"
162+
)
163+
164+
# Check if parent is also a derived view first - if so, recursively resolve to original source
165+
if hasattr(parent_view, "source_views") and parent_view.source_views:
166+
# Parent is also a derived view - recursively find original source
167+
original_source_view = find_original_source_view(parent_view)
168+
return FeatureViewSourceInfo(
169+
data_source=original_source_view.batch_source,
170+
source_type="original_source",
171+
has_transformation=view_has_transformation,
172+
transformation_func=transformation_func,
173+
source_description=f"Original source {original_source_view.name} batch_source for derived view {feature_view.name} (via {parent_view.name})",
174+
)
175+
elif hasattr(parent_view, "batch_source") and parent_view.batch_source:
176+
# Parent has a direct batch_source, use it
177+
return FeatureViewSourceInfo(
178+
data_source=parent_view.batch_source,
179+
source_type="batch_source",
180+
has_transformation=view_has_transformation,
181+
transformation_func=transformation_func,
182+
source_description=f"Parent {parent_view.name} batch_source for derived view {feature_view.name}",
183+
)
184+
else:
185+
# No valid source found
186+
raise ValueError(
187+
f"Unable to resolve data source for derived feature view {feature_view.name} via parent {parent_view.name}"
188+
)
189+
190+
191+
def resolve_feature_view_source_with_fallback(
192+
feature_view: "FeatureView",
193+
config: Optional["RepoConfig"] = None,
194+
is_materialization: bool = False,
195+
) -> FeatureViewSourceInfo:
196+
"""
197+
Resolve feature view source with fallback error handling.
198+
199+
This version includes additional error handling and fallback logic
200+
for cases where the primary resolution fails.
201+
"""
202+
try:
203+
return resolve_feature_view_source(feature_view, config, is_materialization)
204+
except Exception as e:
205+
logger.warning(f"Primary source resolution failed for {feature_view.name}: {e}")
206+
207+
# Fallback: try to find any available source
208+
if hasattr(feature_view, "batch_source") and feature_view.batch_source:
209+
return FeatureViewSourceInfo(
210+
data_source=feature_view.batch_source,
211+
source_type="fallback_batch_source",
212+
has_transformation=has_transformation(feature_view),
213+
transformation_func=get_transformation_function(feature_view),
214+
source_description=f"Fallback batch_source for {feature_view.name}",
215+
)
216+
elif hasattr(feature_view, "source_views") and feature_view.source_views:
217+
# Try the original source view as last resort
218+
original_view = find_original_source_view(feature_view)
219+
return FeatureViewSourceInfo(
220+
data_source=original_view.batch_source,
221+
source_type="fallback_original_source",
222+
has_transformation=has_transformation(feature_view),
223+
transformation_func=get_transformation_function(feature_view),
224+
source_description=f"Fallback original source {original_view.name} for {feature_view.name}",
225+
)
226+
else:
227+
raise ValueError(
228+
f"Unable to resolve any data source for feature view {feature_view.name}"
229+
)

‎sdk/python/feast/infra/compute_engines/ray/compute.py‎

Lines changed: 35 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -199,14 +199,43 @@ def _materialize_from_offline_store(
199199
end_date=end_date,
200200
)
201201

202-
# Convert to Arrow Table and write to online store
202+
# Convert to Arrow Table and write to online/offline stores
203203
arrow_table = retrieval_job.to_arrow()
204-
# TODO: Implement proper online store writing with correct data format conversion
205-
# self.online_store.online_write_batch(...)
206-
logger.debug(
207-
f"Materialization completed, arrow table has {arrow_table.num_rows} rows"
208-
)
209204

205+
# Write to online store if enabled
206+
if getattr(feature_view, "online", False):
207+
# TODO: Implement proper online store writing with correct data format conversion
208+
logger.debug(
209+
f"Online store writing not implemented yet for {arrow_table.num_rows} rows"
210+
)
211+
212+
# Write to offline store if enabled (this handles sink_source automatically for derived views)
213+
if getattr(feature_view, "offline", False):
214+
self.offline_store.offline_write_batch(
215+
config=self.repo_config,
216+
feature_view=feature_view,
217+
table=arrow_table,
218+
progress=lambda x: None,
219+
)
220+
221+
# For derived views, also ensure data is written to sink_source if it exists
222+
# This is critical for feature view chaining to work properly
223+
sink_source = getattr(feature_view, "sink_source", None)
224+
if sink_source is not None:
225+
logger.debug(
226+
f"Writing derived view {feature_view.name} to sink_source: {sink_source.path}"
227+
)
228+
229+
# Write to sink_source using Ray data
230+
try:
231+
# Convert arrow table to pandas then to ray dataset
232+
df = arrow_table.to_pandas()
233+
ray_dataset = ray.data.from_pandas(df)
234+
ray_dataset.write_parquet(sink_source.path)
235+
except Exception as e:
236+
logger.error(
237+
f"Failed to write to sink_source {sink_source.path}: {e}"
238+
)
210239
return RayMaterializationJob(
211240
job_id=job_id,
212241
status=MaterializationJobStatus.SUCCEEDED,

‎sdk/python/feast/infra/compute_engines/ray/config.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
"""Configuration for Ray compute engine."""
22

33
from datetime import timedelta
4-
from typing import Dict, Literal, Optional
4+
from typing import Any, Dict, Literal, Optional
55

66
from pydantic import StrictStr
77

@@ -39,7 +39,7 @@ class RayComputeEngineConfig(FeastConfigBaseModel):
3939
window_size_for_joins: str = "1H"
4040
"""Window size for windowed temporal joins"""
4141

42-
ray_conf: Optional[Dict[str, str]] = None
42+
ray_conf: Optional[Dict[str, Any]] = None
4343
"""Ray configuration parameters"""
4444

4545
# Additional configuration options

0 commit comments

Comments
 (0)