Repository navigation
Expr.over() on an aggregate silently drops order_by, null_treatment, filter, and distinct #1764
Description
Describe the bug
Expr.over() on an aggregate function builds a new window function from only the aggregate's function and arguments. Any order_by, null_treatment, filter, or distinct set on the aggregate is silently dropped, so the window result differs from what the aggregate would compute over the same partition.
To Reproduce
from datafusion import SessionContext, col, functions as F
from datafusion.common import NullTreatment
from datafusion.expr import Window
ctx = SessionContext()
df = ctx.from_pydict({"g": [1, 1, 1], "i": [1, 2, 3], "v": [None, 5, 7]})
cases = {
"first_value": F.first_value(
col("v"),
order_by=col("i").sort(ascending=False),
null_treatment=NullTreatment.IGNORE_NULLS,
),
"sum filter": F.sum(col("v"), filter=col("i") > 2),
"count distinct": F.count(col("g"), distinct=True),
}
for name, e in cases.items():
w = e.over(Window(partition_by=[col("g")]))
print(name, w.canonical_name())
print(" aggregate:", df.aggregate([], [e.alias("r")]).collect_column("r").to_pylist())
print(" window: ", df.select(w.alias("r")).collect_column("r").to_pylist())| Expression | Aggregate | .over(Window(partition_by=[g])) |
|---|---|---|
first_value(v, order_by=i DESC, IGNORE_NULLS) |
7 |
[None, None, None] |
sum(v, filter=i > 2) |
7 |
[12, 12, 12] |
count(g, distinct=True) |
1 |
[3, 3, 3] |
The generated name shows the loss, e.g. first_value(v) PARTITION BY [g] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING, with no ordering or IGNORE NULLS.
Expected behavior
Options set on the aggregate either carry into the window function or raise an error. They should never be silently discarded.
Additional context
The cause is the Expr::AggregateFunction branch of PyExpr::over in crates/core/src/expr.rs, which calls WindowFunction::new(AggregateUDF(agg_fn.func), agg_fn.params.args) and ignores the rest of agg_fn.params. The Expr::WindowFunction branch had the same problem and was fixed in 8173554 by starting from builder_from_expr.
Possible approach:
null_treatment,filter, anddistinctmap directly ontoWindowFunctionParamsand can be carried over. Execution support forfilter/distinctin window aggregates should be confirmed, and an error raised where unsupported.order_byneeds a decision. On an aggregate it orders the values fed to the accumulator. On a window it orders rows and switches the default frame from the whole partition to a running frame. Carrying it over blindly fixesfirst_valuebut turnssum(v, order_by=...)into a running sum, and it conflicts whenWindowalso setsorder_by. Options: raise when the aggregate has anorder_by, or carry it only whenWindowsets neitherorder_bynor a frame and pin the frame to the whole partition.- The
Expr.overdocstring ("how each of the parameters is used is determined by the underlying aggregate function") should state the resulting behavior.
Activity
docs: note that over() drops options set on an aggregate
fix: keep aggregate options in over() and reject an aggregate order_by