def METHOD_NAME(step: WaterfallStep, start_df: DataFrame, end_df: DataFrame) -> DataFrame:
    group_by_columns = [step.labelsColumn] + step.groupby
    if step.parentsColumn is not None:
        group_by_columns = group_by_columns + [step.parentsColumn]
    start_df = (
        start_df.groupby(by=group_by_columns)
        .agg({step.valueColumn + "_start": "sum"})
        .METHOD_NAME(start_df[group_by_columns], on=group_by_columns)
        .rename(columns={step.labelsColumn + "_x": step.labelsColumn})
        .drop_duplicates()
    )
    end_df = (
        end_df.groupby(by=group_by_columns)
        .agg({step.valueColumn + "_end": "sum"})
        .METHOD_NAME(end_df[group_by_columns], on=group_by_columns)
        .rename(columns={step.labelsColumn + "_x": step.labelsColumn})
        .drop_duplicates()
    )
    # we join the result to compare them
    merged_df = start_df.METHOD_NAME(end_df, on=get_join_key(step))
    merged_df[_RESULT_COLUMN] = (
        merged_df[f"{step.valueColumn}_end"] - merged_df[f"{step.valueColumn}_start"]
    )
    merged_df = merged_df.drop(
        columns=[
            f"{step.valueColumn}_start",
            f"{step.valueColumn}_end",
        ]
    )
    # if there is a parent column, we need to aggregate for them
    if step.parentsColumn is not None:
        parents_results = merged_df.groupby(
            step.groupby + [step.parentsColumn], as_index=False
        ).agg({_RESULT_COLUMN: "sum"})
        parents_results[step.labelsColumn] = parents_results[step.parentsColumn]
        return pd.concat([merged_df, parents_results])
    return merged_df