Expr.over() on an aggregate silently drops order_by, null_treatment, filter, and distinct

[timsaucer]

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, and distinct map directly onto WindowFunctionParams and can be carried over. Execution support for filter / distinct in window aggregates should be confirmed, and an error raised where unsupported.
• order_by needs 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 fixes first_value but turns sum(v, order_by=...) into a running sum, and it conflicts when Window also sets order_by. Options: raise when the aggregate has an order_by, or carry it only when Window sets neither order_by nor a frame and pin the frame to the whole partition.
• The Expr.over docstring ("how each of the parameters is used is determined by the underlying aggregate function") should state the resulting behavior.