Add Chaining Reusable Transformation Functions part of Feature Views #4696

@transformation(mode="big_query")
def last_balance_time(transactions, max_user_transaction):
    return f"""SELECT t.user_id, t.current_balance, last_t.last_transaction_date as timestamp
                      FROM {transactions} t
                      INNER JOIN {max_user_transaction} last_t
                      ON t.user_id = last_t.user_id AND t.timestamp = last_t.last_transaction_date;"""

@transformation(mode="big_query")
def user_last_transaction_time(transactions):
    return f"""SELECT user_id, MAX(timestamp) AS last_transaction_date
                      FROM {transactions}
                      GROUP BY user_id"""

@feature_view(
    sources=[credit_data_batch],
    entities=[user],
    mode="pipeline", # creates a DAG from re-useable transformation functions
    batch_schedule=timedelta(days=1),
    schema=[Field("user_id", String), Field("timestamp", Timestamp), Field("current_balance", Float64)],
)
def user_last_balance(transactions):
    user_last_transaction_time = user_last_transaction_time(transactions)
    return last_balance_time(transactions, user_last_transaction_time)