update doc

HaoXuAI · HaoXuAI · commit 95f757d54448 · 2025-04-12T23:07:18.000-07:00
Signed-off-by: HaoXuAI &lt;sduxuhao@gmail.com&gt;
diff --git a/docs/reference/compute-engine/README.md b/docs/reference/compute-engine/README.md
@@ -38,8 +38,33 @@ This system builds and executes DAGs (Directed Acyclic Graphs) of typed operatio
 
 ---
 
-## 🛠️ Example DAG Flow
-`Read → Aggregate → Join → Transform → Write`
+## 🛠️ Feature Builder Flow 
+```markdown
+SourceReadNode
+      |
+      v
+JoinNode (Only for get_historical_features with entity df)
+      |
+      v
+FilterNode (Always included; applies TTL or user-defined filters)
+      |
+      v
+AggregationNode (If aggregations are defined in FeatureView)
+      |
+      v
+DeduplicationNode (If no aggregation is defined for get_historical_features) 
+      |
+      v
+TransformationNode (If feature_transformation is defined)
+      |
+      v
+ValidationNode (If enable_validation = True)
+      |
+      v
+Output
+  ├──> RetrievalOutput (For get_historical_features)
+  └──> OnlineStoreWrite / OfflineStoreWrite (For materialize)
+```
 
 Each step is implemented as a `DAGNode`. An `ExecutionPlan` executes these nodes in topological order, caching `DAGValue` outputs.
 
@@ -59,7 +84,7 @@ class MyComputeEngine(ComputeEngine):
     def materialize(self, task: MaterializationTask) -> MaterializationJob:
         ...
 
-    def get_historical_features(self, task: HistoricalRetrievalTask) -> pa.Table:
+    def get_historical_features(self, task: HistoricalRetrievalTask) -> RetrievalJob:
         ...
 ```
 
@@ -71,6 +96,8 @@ class CustomFeatureBuilder(FeatureBuilder):
     def build_source_node(self): ...
     def build_aggregation_node(self, input_node): ...
     def build_join_node(self, input_node): ...
+    def build_filter_node(self, input_node):
+    def build_dedup_node(self, input_node):
     def build_transformation_node(self, input_node): ...
     def build_output_nodes(self, input_node): ...
 ```
diff --git a/sdk/python/feast/infra/compute_engines/spark/compute.py b/sdk/python/feast/infra/compute_engines/spark/compute.py
@@ -54,7 +54,6 @@ def materialize(self, task: MaterializationTask) -> MaterializationJob:
             )
 
         except Exception as e:
-            raise e
             # 🛑 Handle failure
             return SparkMaterializationJob(
                 job_id=job_id, status=MaterializationJobStatus.ERROR, error=e

Original file line number	Diff line number	Diff line change
`@@ -54,7 +54,6 @@ def materialize(self, task: MaterializationTask) -> MaterializationJob:`
`54`	`54`	`)`
`55`	`55`
`56`	`56`	`except Exception as e:`
`57`		`- raise e`
`58`	`57`	`# 🛑 Handle failure`
`59`	`58`	`return SparkMaterializationJob(`
`60`	`59`	`job_id=job_id, status=MaterializationJobStatus.ERROR, error=e`