GeezerVison is showing github.com — fix: Stop the Ray engine from hiding materialization errors by patelchaitany · Pull Request #7026 · feast-dev/feast
fix: Stop the Ray engine from hiding materialization errors
#7026
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.
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
Uh oh!
There was an error while loading. Please reload this page.
Show file tree
Hide file tree
Changes from all commits
Commits
File filter
Filter by extension
.md
(1)
.py
(5)
All 2 file types selected
Viewed files
Conversations
Failed to load comments.
Loading
Uh oh!
There was an error while loading. Please reload this page.
Jump to
Jump to file
Failed to load files.
Loading
Uh oh!
There was an error while loading. Please reload this page.
Diff view
Hide whitespace
Diff view
Hide whitespace
There are no files selected for viewing
Original file line number
Diff line number
Diff line change
Expand Up
@@ -35,7 +35,7 @@ The Ray compute engine provides:
The Ray compute engine follows Feast's DAG-based architecture:
```
EntityDF → RayReadNode → RayJoinNode → RayFilterNode → RayAggregationNode → RayTransformationNode → Output
EntityDF → RayReadNode → RayTransformationNode → RayJoinNode → RayFilterNode → RayAggregationNode → Output
```
### Core Components
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -88,6 +88,9 @@ def build_aggregation_node(self, view, input_node: DAGNode) -> DAGNode:
config=self.config,
enable_tiling=enable_tiling,
hop_size=tiling_hop_size,
created_timestamp_col=getattr(
view.batch_source, "created_timestamp_column", None
),
)
node.add_input(input_node)
Expand Down
Expand Up
@@ -333,19 +336,19 @@ def _build_aggregation_optimized_plan(self) -> ExecutionPlan:
# 1. Read source data
last_node = self.build_source_node(self.feature_view)
# 2. Apply filters (TTL, custom filters) BEFORE aggregation
# 2. Apply transformations to the raw source data, as materialization does
Comment thread
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more .
is this order change matching other compute engines?
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more .
Yes. Local, Spark and Flink all run the UDF before aggregation (shared FeatureBuilder._build
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
if self._should_transform(self.feature_view):
last_node = self.build_transformation_node(self.feature_view, [last_node])
# 3. Apply filters (TTL, custom filters) BEFORE aggregation
last_node = self.build_filter_node(self.feature_view, last_node)
# 3 . Aggregate across all historical records
# 4 . Aggregate across all historical records
last_node = self.build_aggregation_node(self.feature_view, last_node)
# 4 . Join with entity_df to get aggregated features for each entity
# 5 . Join with entity_df to get aggregated features for each entity
last_node = self.build_join_node(self.feature_view, [last_node])
# 5. Apply transformations to aggregated features
if self._should_transform(self.feature_view):
last_node = self.build_transformation_node(self.feature_view, [last_node])
# 6. Output
last_node = self.build_output_nodes(self.feature_view, last_node)
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -20,6 +20,7 @@
from feast.infra.compute_engines.dag.value import DAGValue
from feast.infra.compute_engines.ray.config import RayComputeEngineConfig
from feast.infra.compute_engines.ray.utils import (
_is_empty_batch,
safe_batch_processor,
write_to_online_store,
)
Expand Down
Expand Up
@@ -329,11 +330,13 @@ def __init__(
config: RayComputeEngineConfig,
enable_tiling: bool = False,
hop_size: Optional[timedelta] = None,
created_timestamp_col: Optional[str] = None,
):
super().__init__(name)
self.aggregations = aggregations
self.group_by_keys = group_by_keys
self.timestamp_col = timestamp_col
self.created_timestamp_col = created_timestamp_col
self.config = config
self.enable_tiling = enable_tiling
self.hop_size = hop_size
Expand Down
Expand Up
@@ -443,6 +446,21 @@ def _execute_tiled_aggregation(self, dataset: Dataset) -> DAGValue:
final_pdf, self.group_by_keys, self.timestamp_col
)
# Tiling drops the created timestamp; add back the latest per group.
if (
self.created_timestamp_col
and self.created_timestamp_col in input_pdf.columns
and not final_pdf.empty
):
latest_created = (
input_pdf.groupby(self.group_by_keys)[self.created_timestamp_col]
.max()
.reset_index()
)
final_pdf = final_pdf.merge(
latest_created, on=self.group_by_keys, how="left"
)
aggregated_dataset = ray_wrapper.from_pandas(final_pdf)
return DAGValue(
Expand Down
Expand Up
@@ -543,6 +561,10 @@ def _fallback_pandas_aggregation(self, dataset: Dataset, agg_dict: dict) -> Data
# Combine aggregation results
if agg_results:
# Keep the latest timestamps per group; the online write needs them.
Comment thread
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more .
Tiled aggregation doesn’t preserve created_timestamp_col (fix only touched the standard _fallback_pandas_aggregation).
Sorry, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
patelchaitany reacted with thumbs up emoji
for ts_col in (self.timestamp_col, self.created_timestamp_col):
if ts_col and ts_col in df.columns:
agg_results.append(grouped[ts_col].max())
result_df = pd.concat(agg_results, axis=1)
# Reset index to make group keys regular columns
Expand Down
Expand Up
@@ -708,11 +730,14 @@ def execute(self, context: ExecutionContext) -> DAGValue:
elif callable(self.transformation):
transformation_serialized = dill.dumps(self.transformation)
@ safe_batch_processor
# Not wrapped in safe_batch_processor: a failed UDF must fail the job.
def apply_transformation_with_serialized_udf(
batch: pd.DataFrame,
) -> pd.DataFrame:
"""Apply the transformation using pre-serialized UDF."""
if _is_empty_batch(batch):
return batch
if transformation_serialized:
transformation_func = dill.loads(transformation_serialized)
transformed_batch = transformation_func(batch)
Expand Down
Expand Up
@@ -861,9 +886,12 @@ def execute(self, context: ExecutionContext) -> DAGValue:
feature_view=self.feature_view, repo_config=context.repo_config
)
@ safe_batch_processor
# Not wrapped in safe_batch_processor: a failed write must fail the job.
def write_batch_with_serialized_artifacts(batch: pd.DataFrame) -> pd.DataFrame:
"""Write each batch using pre-serialized artifacts."""
if _is_empty_batch(batch):
return batch
(
feature_view,
online_store,
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -81,6 +81,7 @@ def write_to_online_store(
except Exception as e:
logger.error(f"Failed to write to online store for {feature_view.name}: {e}")
raise
# Ray Data batch type: pandas DataFrame, numpy dict, or pyarrow Table
Expand Down
Expand Up
@@ -178,6 +179,7 @@ def _write_batch(batch: pa.Table) -> pa.Table:
logger.error(
f"Distributed write to online store failed for {feature_view.name}: {e}"
)
raise
def safe_batch_processor(
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -41,9 +41,10 @@ def test_ray_compute_engine_get_historical_features(
fs = ray_environment.feature_store
registry = fs.registry
# The udf runs on raw source rows, before aggregation.
def transform_feature(df: pd.DataFrame) -> pd.DataFrame:
df["sum_conv_rate "] = df["sum_conv_rate "] * 2
df["avg_acc_rate "] = df["avg_acc_rate "] * 2
df["conv_rate "] = df["conv_rate "] * 2
df["acc_rate "] = df["acc_rate "] * 2
return df
driver_stats_fv = BatchFeatureView(
Expand Down
Expand Up
@@ -102,11 +103,6 @@ def test_ray_compute_engine_materialize(ray_environment, feature_dataset):
fs = ray_environment.feature_store
registry = fs.registry
def transform_feature(df: pd.DataFrame) -> pd.DataFrame:
df["sum_conv_rate"] = df["sum_conv_rate"] * 2
df["avg_acc_rate"] = df["avg_acc_rate"] * 2
return df
driver_stats_fv = BatchFeatureView(
name="driver_hourly_stats",
entities=[driver],
Expand All
@@ -115,13 +111,10 @@ def transform_feature(df: pd.DataFrame) -> pd.DataFrame:
Aggregation(column="conv_rate", function="sum"),
Aggregation(column="acc_rate", function="avg"),
],
udf=transform_feature,
udf_string="transform_feature",
ttl=timedelta(days=3),
schema=[
Field(name="conv_rate", dtype=Float32),
Field(name="acc_rate", dtype=Float32),
Field(name="avg_daily_trips", dtype=Int64),
Field(name="sum_conv_rate", dtype=Float32),
Field(name="avg_acc_rate", dtype=Float32),
Field(name="driver_id", dtype=Int32),
],
online=True,
Expand Down
Expand Up
@@ -153,6 +146,17 @@ def tqdm_builder(length):
assert len(ray_materialize_jobs) == 1
assert ray_materialize_jobs[0].status() == MaterializationJobStatus.SUCCEEDED
# driver 1001 has one row in the window: conv_rate 0.8, acc_rate 0.5
online = fs.get_online_features(
features=[
"driver_hourly_stats:sum_conv_rate",
"driver_hourly_stats:avg_acc_rate",
],
entity_rows=[{"driver_id": 1001}],
).to_dict()
assert online["sum_conv_rate"][0] == pytest.approx(0.8)
assert online["avg_acc_rate"][0] == pytest.approx(0.5)
@pytest.mark.integration
@pytest.mark.xdist_group(name="ray")
Expand Down
Original file line number
Diff line number
Diff line change
Expand Up
@@ -6,6 +6,7 @@
import pytest
import ray
from feast import Entity, FeatureView, Field, FileSource, RepoConfig
from feast.aggregation import Aggregation
from feast.infra.compute_engines.dag.context import ColumnInfo
from feast.infra.compute_engines.dag.model import DAGFormat
Expand All
@@ -19,13 +20,16 @@
RayJoinNode,
RayReadNode,
RayTransformationNode,
RayWriteNode,
)
from feast.infra.online_stores.sqlite import SqliteOnlineStoreConfig
from feast.infra.ray_initializer import (
RayConfigManager,
RayExecutionMode,
ensure_ray_initialized,
get_ray_wrapper,
)
from feast.types import Float32
class DummyInputNode(DAGNode):
Expand Down
Expand Up
@@ -195,6 +199,34 @@ def test_ray_aggregation_node(
assert "avg_acc_rate" in result_df.columns
def test_ray_tiled_aggregation_node_keeps_created_timestamp(
ray_session, ray_config, mock_context, sample_data
):
"""Tiled aggregation must keep the latest created timestamp per entity."""
input_value = DAGValue(data=ray.data.from_pandas(sample_data), format=DAGFormat.RAY)
node = RayAggregationNode(
name="aggregation",
aggregations=[
Aggregation(
column="conv_rate", function="sum", time_window=timedelta(days=1)
)
],
group_by_keys=["driver_id"],
timestamp_col="event_timestamp",
config=ray_config,
enable_tiling=True,
hop_size=timedelta(hours=1),
created_timestamp_col="created",
)
node.add_input(DummyInputNode("input_node", input_value))
mock_context.node_outputs = {"input_node": input_value}
result_df = node.execute(mock_context).data.to_pandas().set_index("driver_id")
expected = sample_data.groupby("driver_id")["created"].max()
assert (result_df["created"] == expected.loc[result_df.index]).all()
def test_ray_join_node(ray_session, ray_config, mock_context, sample_data, column_info):
"""Test RayJoinNode functionality."""
entity_data = pd.DataFrame(
Expand Down
Expand Up
@@ -439,6 +471,38 @@ def test_ray_dedup_node_materialization_no_join_keys(
)
def test_ray_write_node_raises_when_online_write_fails(
ray_session, mock_context, sample_data, tmp_path
):
"""A failed online store write must fail the write node, not be swallowed."""
driver = Entity(name="driver", join_keys=["driver_id"])
feature_view = FeatureView(
name="driver_stats",
entities=[driver],
schema=[Field(name="conv_rate", dtype=Float32)],
source=FileSource(
path=str(tmp_path / "source.parquet"), timestamp_field="event_timestamp"
),
online=True,
offline=False,
)
# The online table was never created, so every write raises "no such table".
mock_context.repo_config = RepoConfig(
project="test_project",
registry=str(tmp_path / "registry.db"),
provider="local",
entity_key_serialization_version=3,
online_store=SqliteOnlineStoreConfig(path=str(tmp_path / "online.db")),
)
input_value = DAGValue(data=ray.data.from_pandas(sample_data), format=DAGFormat.RAY)
node = RayWriteNode(name="write", feature_view=feature_view)
node.add_input(DummyInputNode("input_node", input_value))
mock_context.node_outputs = {"input_node": input_value}
with pytest.raises(Exception, match="no such table"):
node.execute(mock_context)
def test_ray_config_validation():
"""Test Ray configuration validation."""
# Test valid configuration
Expand Down
Loading
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.