

@dispatch(By, sa.Column)
def compute_up(expr, data, scope=None, **kwargs):
    data = lower_column(data)
    grouper = compute(expr.grouper, scope, post_compute=False, return_type='native', **kwargs)
    app = expr.apply
    reductions = [compute(val, data, post_compute=None, return_type='native').label(name) for (val, name) in zip(app.values, app.fields)]
    froms = list(unique(chain(get_all_froms(grouper), concat(map(get_all_froms, reductions)))))
    inner_cols = list(getattr(grouper, 'inner_columns', [grouper]))
    grouper_cols = inner_cols[:]
    inner_cols.extend(concat((getattr(getattr(r, 'element', None), 'inner_columns', [r]) for r in reductions)))
    wheres = unify_wheres(([grouper] + reductions))
    sel = unify_froms(sa.select(inner_cols, whereclause=wheres), froms)
    return sel.group_by(*grouper_cols)

dataset/ETHPy150Open blaze/blaze/blaze/compute/sql.py compute_up/original