Devin feedback

Signed-off-by: Nick Quinn <nicholas_quinn@apple.com>

Author: nickquinn408

sdk/python/feast/aggregation/__init__.py

@@ -103,6 +103,20 @@ def __eq__(self, other):
 
         return True
 
+    def resolved_name(self, time_window: Optional[timedelta] = None) -> str:
+        """Return the output feature name for this aggregation.
+
+        If ``name`` is set it is returned as-is.  Otherwise the name is
+        derived as ``{function}_{column}``, with ``_{seconds}s`` appended
+        when *time_window* is provided.
+        """
+        if self.name:
+            return self.name
+        base = f"{self.function}_{self.column}"
+        if time_window is not None:
+            return f"{base}_{int(time_window.total_seconds())}s"
+        return base
+
 
 def aggregation_specs_to_agg_ops(
     agg_specs: Iterable[Any],

sdk/python/feast/aggregation/tiling/orchestrator.py

@@ -61,9 +61,7 @@ def apply_sawtooth_window_tiling(
     ir_metadata_dict = {}
 
     for agg in aggregations:
-        feature_name = (
-            f"{agg.function}_{agg.column}_{int(window_size.total_seconds())}s"
-        )
+        feature_name = agg.resolved_name(window_size)
         _, metadata = get_ir_metadata_for_aggregation(agg, feature_name)
         ir_metadata_dict[feature_name] = metadata
 
@@ -161,9 +159,7 @@ def apply_sawtooth_window_tiling(
 
     # Step 5: Compute final feature values from IRs (for algebraic aggs, just rename)
     for agg in aggregations:
-        feature_name = (
-            f"{agg.function}_{agg.column}_{int(window_size.total_seconds())}s"
-        )
+        feature_name = agg.resolved_name(window_size)
         metadata = ir_metadata_dict[feature_name]
 
         if metadata.type == "algebraic":

sdk/python/feast/aggregation/tiling/tile_subtraction.py

@@ -76,9 +76,7 @@ def convert_cumulative_to_windowed(
 
             # Subtract previous tile values from current tile for each aggregation
             for agg in aggregations:
-                feature_name = (
-                    f"{agg.function}_{agg.column}_{int(window_size.total_seconds())}s"
-                )
+                feature_name = agg.resolved_name(window_size)
                 _, metadata = get_ir_metadata_for_aggregation(agg, feature_name)
 
                 if metadata.type == "algebraic":

sdk/python/feast/infra/compute_engines/ray/nodes.py

@@ -461,9 +461,7 @@ def _execute_standard_aggregation(self, dataset: Dataset) -> DAGValue:
         # Convert aggregations to Ray's groupby format
         agg_dict = {}
         for agg in self.aggregations:
-            feature_name = f"{agg.function}_{agg.column}"
-            if agg.time_window:
-                feature_name += f"_{int(agg.time_window.total_seconds())}s"
+            feature_name = agg.resolved_name(agg.time_window)
 
             if agg.function == "count":
                 agg_dict[feature_name] = (agg.column, "count")

sdk/python/feast/infra/compute_engines/spark/nodes.py

@@ -311,7 +311,7 @@ def _execute_tiled_aggregation(self, input_df: DataFrame) -> DAGValue:
             expected_columns = entity_keys + [self.timestamp_col]
             for time_window, window_aggs in aggs_by_window.items():
                 for agg in window_aggs:
-                    feature_name = f"{agg.function}_{agg.column}_{int(time_window.total_seconds())}s"
+                    feature_name = agg.resolved_name(time_window)
                     if feature_name not in expected_columns:
                         expected_columns.append(feature_name)
 
@@ -372,11 +372,7 @@ def _execute_standard_aggregation(self, input_df: DataFrame) -> DAGValue:
         agg_exprs = []
         for agg in self.aggregations:
             func = getattr(F, agg.function)
-            expr = func(agg.column).alias(
-                f"{agg.function}_{agg.column}_{int(agg.time_window.total_seconds())}s"
-                if agg.time_window
-                else f"{agg.function}_{agg.column}"
-            )
+            expr = func(agg.column).alias(agg.resolved_name(agg.time_window))
             agg_exprs.append(expr)
 
         if any(agg.time_window for agg in self.aggregations):