@@ -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
 
@@ -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)
 
@@ -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
+        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)
 
 
@@ -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,
 )
@@ -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
@@ -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(
@@ -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.
+            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
@@ -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)
@@ -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,
 
@@ -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
@@ -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(
 
@@ -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(
@@ -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],
@@ -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,
@@ -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")
 
@@ -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
@@ -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):
@@ -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(
@@ -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