Visitar URL original
feat: Adding optional name to Aggregation (feast-dev#5994) by nquinn408 · Pull Request #6083 · feast-dev/feast · GitHub
Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Devin feedback
Signed-off-by: Nick Quinn <nicholas_quinn@apple.com>
  • Loading branch information
nickquinn408 committed Mar 9, 2026
commit cf0a2745a6ec1a7f21fc3d35d529ca34ac61771e
14 changes: 14 additions & 0 deletions sdk/python/feast/aggregation/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Comment thread
nquinn408 marked this conversation as resolved.


def aggregation_specs_to_agg_ops(
agg_specs: Iterable[Any],
Expand Down
8 changes: 2 additions & 6 deletions sdk/python/feast/aggregation/tiling/orchestrator.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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":
Expand Down
4 changes: 1 addition & 3 deletions sdk/python/feast/aggregation/tiling/tile_subtraction.py
Original file line number Diff line number Diff line change
Expand Up @@ -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":
Expand Down
4 changes: 1 addition & 3 deletions sdk/python/feast/infra/compute_engines/ray/nodes.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
8 changes: 2 additions & 6 deletions sdk/python/feast/infra/compute_engines/spark/nodes.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down Expand Up @@ -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):
Expand Down
Loading