Record whether a window frame was set explicitly, so ExprFunctionExt can build on an existing window function #25934

Is your feature request related to a problem or challenge?

ExprFunctionExt on an existing Expr starts from an empty builder. ExprFuncBuilder::new sets every option to None, and build() then overwrites all of the function's params. So chaining one more option onto a function that already has options silently resets the others:

let e = sum(col("v"))
    .order_by(vec![col("t").sort(true, false)])
    .filter(col("v").gt(lit(0)))
    .build()?;

// Only DISTINCT survives: ORDER BY and FILTER are reset.
let e2 = e.distinct().build()?;

Bindings that expose both keyword arguments and a builder chain hit this all the time. In datafusion-python, f.string_agg(s, ",", order_by=v).distinct() silently drops the ordering, and f.lead(v, order_by=t).over(Window(partition_by=[g])) drops the ORDER BY (apache/datafusion-python#1764 is the aggregate case).

For aggregates, the fix is mechanical: seed the builder from AggregateFunctionParams, since every field there is stored exactly as the user set it.

For window functions, it can't be done correctly today, because WindowFunctionParams.window_frame doesn't say whether the frame was chosen or derived. build() fills in WindowFrame::new(has_order_by) when no frame was set, and the SQL planner does the same when the OVER clause has no frame. After that, these two are identical:

// The user asked for the whole partition.
let a = first_value(col("v")).window_frame(WindowFrame::new(None)).build()?;
// The user asked for nothing.
let b = first_value(col("v")).build()?;
assert_eq!(a, b);

If a later call adds ORDER BY, b should switch to the default running frame (RANGE UNBOUNDED PRECEDING .. CURRENT ROW), and a should keep the whole partition. Code that seeds a builder from a or b can't tell which one it has, so any rule it picks silently gives wrong results for one of them.

A downstream workaround also has to undo regularize_order_bys. Decoding (proto and SQL) gives a free RANGE frame with no ORDER BY the sort key UInt64(1). A seeding rule then has to recognize that key as "no ORDER BY" when deciding whether the frame is a default. But it must still keep the key whenever it keeps the frame: dropping the key while keeping the RANGE frame gives an expression that fails with ORDER BY column cannot be empty.

In datafusion-python, a workaround that compares the stored frame against WindowFrame::new(..) has needed a series of fixes for these cases, and still can't keep a user-chosen frame that happens to equal the default. Our current plan is to raise an error when chaining onto a window function that already has window options, until the expression can carry this information.

Describe the solution you'd like

Record on the expression whether the frame was explicit. For example, add a field to WindowFunctionParams:

pub struct WindowFunctionParams {
    // ...
    pub window_frame: WindowFrame,
    /// Whether `window_frame` was set explicitly, rather than derived from `order_by`.
    pub window_frame_explicit: bool,
}

Set it everywhere a window function is created:

• ExprFuncBuilder::build(): true when window_frame(..) was called.
• The SQL planner: true when the OVER clause has a frame clause.
• datafusion-proto: a new bool field on WindowExprNode. Proto3 defaults it to false, so plans serialized before the change decode as "derived", which matches how they are treated today.
• Substrait: whatever it can represent. false when the frame bounds are absent.
• WindowFunction::new and other constructors: false.

With the flag in place, impl ExprFunctionExt for Expr can seed the builder from the existing params and leave window_frame as None when the frame wasn't explicit. build() then derives it from the final ORDER BY, so chaining behaves like setting every option in one call. An explicit frame and its regularize_order_bys key are kept together. A new order_by(..) replaces the key.

The flag should probably not take part in PartialEq/Hash, so that OVER () and OVER (ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) still compare equal for common-subexpression elimination and plan comparison. That needs hand-written impls for WindowFunctionParams. Feedback welcome on whether the flag should instead take part in equality.

Describe alternatives you've considered

• Store Option<WindowFrame> and resolve it at planning time. This is the most faithful model, but every consumer of window_frame in the optimizer and physical planner would have to handle None. A flag next to the resolved frame keeps existing readers unchanged.
• Infer it: treat a frame equal to WindowFrame::new(..) as unset. This is what downstream does today. It silently re-derives a frame the user set explicitly to the default value, and it has to special-case the regularize_order_bys key.
• Keep the reset semantics and document them. Chaining onto a built function would remain a silent way to drop options.

This differs from #12195, which removed Option from order_by because an empty list means the same as no list. For frames, "not set" and "set to the default" behave differently as soon as ORDER BY changes, so the difference has to be recorded somewhere.

Additional context

• partition_by and filter on ExprFunctionExt #21697 (and the stale PR fix: handle WindowFunction in ExprFunctionExt filter and distinct on Expr #21920) covers a related gap in impl ExprFunctionExt for Expr: filter and distinct only handle AggregateFunction, and partition_by only handles WindowFunction, so call order decides whether a chain errors. Seeding the builder from the existing expression would fix that too.
• build() computes has_order_by as order_by.as_ref().map(|o| !o.is_empty()), so order_by(vec![]) gives Some(false), and WindowFrame::new(Some(false)) is RANGE UNBOUNDED PRECEDING .. CURRENT ROW with no sort key. That expression fails at execution with ORDER BY column cannot be empty. Treating an empty list as None here would match Convert Option<Vec<sort expression>> to Vec<sort expression> #12195.
• Downstream context: Expr.over() on an aggregate silently drops order_by, null_treatment, filter, and distinct datafusion-python#1764 and the datafusion-python branch that tried the inference approach.