bug: UDAF can not be used as window function

[rohankrao]

Describe the bug

There is an error while using udaf as a window function with order by in the image nightly-20240905

Error message/log

2024-09-06T07:13:42.267863829Z ERROR   rw-standalone-compute risingwave_batch::executor: failed to build executor error=Expr error: expect user defined plan_node=Ok(SortOverWindow(SortOverWindowNode { calls: [WindowFunction { args: [InputRef { index: 1, r#type: Some(INT32) }, InputRef { index: 0, r#type: Some(INT32) }], return_type: Some(DOUBLE), frame: Some(WindowFrame { r#type: Range, start: None, end: None, exclusion: NoOthers, bounds: Some(Range(RangeFrameBounds { start: Some(RangeFrameBound { r#type: Preceding, offset: Some(Datum { body: [1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 96, 215, 29, 20, 0, 0, 0] }) }), end: Some(RangeFrameBound { r#type: CurrentRow, offset: None }), order_data_type: Some(TIMESTAMP), order_type: Some(OrderType { direction: Descending, nulls_are: Largest }), offset_data_type: Some(INTERVAL) })) }), r#type: Some(Aggregate(UserDefined)) }], partition_by: [2], order_by: [ColumnOrder { column_index: 4, order_type: Some(OrderType { direction: Descending, nulls_are: Largest }) }] }))
2024-09-06T07:13:42.268018519Z ERROR   rw-standalone-compute risingwave_batch::executor: failed to build executor error=Expr error: expect user defined plan_node=Ok(SortOverWindow(SortOverWindowNode { calls: [WindowFunction { args: [InputRef { index: 1, r#type: Some(INT32) }, InputRef { index: 0, r#type: Some(INT32) }], return_type: Some(DOUBLE), frame: Some(WindowFrame { r#type: Range, start: None, end: None, exclusion: NoOthers, bounds: Some(Range(RangeFrameBounds { start: Some(RangeFrameBound { r#type: Preceding, offset: Some(Datum { body: [1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 96, 215, 29, 20, 0, 0, 0] }) }), end: Some(RangeFrameBound { r#type: CurrentRow, offset: None }), order_data_type: Some(TIMESTAMP), order_type: Some(OrderType { direction: Descending, nulls_are: Largest }), offset_data_type: Some(INTERVAL) })) }), r#type: Some(Aggregate(UserDefined)) }], partition_by: [2], order_by: [ColumnOrder { column_index: 4, order_type: Some(OrderType { direction: Descending, nulls_are: Largest }) }] }))
2024-09-06T07:13:42.268026349Z ERROR   rw-standalone-compute risingwave_batch::executor: failed to build executor error=Expr error: expect user defined plan_node=Ok(SortOverWindow(SortOverWindowNode { calls: [WindowFunction { args: [InputRef { index: 1, r#type: Some(INT32) }, InputRef { index: 0, r#type: Some(INT32) }], return_type: Some(DOUBLE), frame: Some(WindowFrame { r#type: Range, start: None, end: None, exclusion: NoOthers, bounds: Some(Range(RangeFrameBounds { start: Some(RangeFrameBound { r#type: Preceding, offset: Some(Datum { body: [1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 96, 215, 29, 20, 0, 0, 0] }) }), end: Some(RangeFrameBound { r#type: CurrentRow, offset: None }), order_data_type: Some(TIMESTAMP), order_type: Some(OrderType { direction: Descending, nulls_are: Largest }), offset_data_type: Some(INTERVAL) })) }), r#type: Some(Aggregate(UserDefined)) }], partition_by: [2], order_by: [ColumnOrder { column_index: 4, order_type: Some(OrderType { direction: Descending, nulls_are: Largest }) }] }))
2024-09-06T07:13:42.268136089Z ERROR   rw-standalone-compute risingwave_batch::rpc::service::task_service: failed to fire task error=failed to build executor: Expr error: expect user defined
2024-09-06T07:13:42.268134849Z ERROR   rw-standalone-compute risingwave_batch::executor: failed to build executor error=Expr error: expect user defined plan_node=Ok(SortOverWindow(SortOverWindowNode { calls: [WindowFunction { args: [InputRef { index: 1, r#type: Some(INT32) }, InputRef { index: 0, r#type: Some(INT32) }], return_type: Some(DOUBLE), frame: Some(WindowFrame { r#type: Range, start: None, end: None, exclusion: NoOthers, bounds: Some(Range(RangeFrameBounds { start: Some(RangeFrameBound { r#type: Preceding, offset: Some(Datum { body: [1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 96, 215, 29, 20, 0, 0, 0] }) }), end: Some(RangeFrameBound { r#type: CurrentRow, offset: None }), order_data_type: Some(TIMESTAMP), order_type: Some(OrderType { direction: Descending, nulls_are: Largest }), offset_data_type: Some(INTERVAL) })) }), r#type: Some(Aggregate(UserDefined)) }], partition_by: [2], order_by: [ColumnOrder { column_index: 4, order_type: Some(OrderType { direction: Descending, nulls_are: Largest }) }] }))
2024-09-06T07:13:42.268034079Z ERROR   rw-standalone-compute risingwave_batch::executor: failed to build executor error=Expr error: expect user defined plan_node=Ok(SortOverWindow(SortOverWindowNode { calls: [WindowFunction { args: [InputRef { index: 1, r#type: Some(INT32) }, InputRef { index: 0, r#type: Some(INT32) }], return_type: Some(DOUBLE), frame: Some(WindowFrame { r#type: Range, start: None, end: None, exclusion: NoOthers, bounds: Some(Range(RangeFrameBounds { start: Some(RangeFrameBound { r#type: Preceding, offset: Some(Datum { body: [1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 96, 215, 29, 20, 0, 0, 0] }) }), end: Some(RangeFrameBound { r#type: CurrentRow, offset: None }), order_data_type: Some(TIMESTAMP), order_type: Some(OrderType { direction: Descending, nulls_are: Largest }), offset_data_type: Some(INTERVAL) })) }), r#type: Some(Aggregate(UserDefined)) }], partition_by: [2], order_by: [ColumnOrder { column_index: 4, order_type: Some(OrderType { direction: Descending, nulls_are: Largest }) }] }))
2024-09-06T07:13:42.268140189Z ERROR   rw-standalone-compute risingwave_batch::executor: failed to build executor error=Expr error: expect user defined plan_node=Ok(SortOverWindow(SortOverWindowNode { calls: [WindowFunction { args: [InputRef { index: 1, r#type: Some(INT32) }, InputRef { index: 0, r#type: Some(INT32) }], return_type: Some(DOUBLE), frame: Some(WindowFrame { r#type: Range, start: None, end: None, exclusion: NoOthers, bounds: Some(Range(RangeFrameBounds { start: Some(RangeFrameBound { r#type: Preceding, offset: Some(Datum { body: [1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 96, 215, 29, 20, 0, 0, 0] }) }), end: Some(RangeFrameBound { r#type: CurrentRow, offset: None }), order_data_type: Some(TIMESTAMP), order_type: Some(OrderType { direction: Descending, nulls_are: Largest }), offset_data_type: Some(INTERVAL) })) }), r#type: Some(Aggregate(UserDefined)) }], partition_by: [2], order_by: [ColumnOrder { column_index: 4, order_type: Some(OrderType { direction: Descending, nulls_are: Largest }) }] }))
2024-09-06T07:13:42.268226849Z ERROR   rw-standalone-compute risingwave_batch::executor: failed to build executor error=Expr error: expect user defined plan_node=Ok(SortOverWindow(SortOverWindowNode { calls: [WindowFunction { args: [InputRef { index: 1, r#type: Some(INT32) }, InputRef { index: 0, r#type: Some(INT32) }], return_type: Some(DOUBLE), frame: Some(WindowFrame { r#type: Range, start: None, end: None, exclusion: NoOthers, bounds: Some(Range(RangeFrameBounds { start: Some(RangeFrameBound { r#type: Preceding, offset: Some(Datum { body: [1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 96, 215, 29, 20, 0, 0, 0] }) }), end: Some(RangeFrameBound { r#type: CurrentRow, offset: None }), order_data_type: Some(TIMESTAMP), order_type: Some(OrderType { direction: Descending, nulls_are: Largest }), offset_data_type: Some(INTERVAL) })) }), r#type: Some(Aggregate(UserDefined)) }], partition_by: [2], order_by: [ColumnOrder { column_index: 4, order_type: Some(OrderType { direction: Descending, nulls_are: Largest }) }] }))
2024-09-06T07:13:42.268190819Z ERROR   rw-standalone-compute risingwave_batch::rpc::service::task_service: failed to fire task error=failed to build executor: Expr error: expect user defined
2024-09-06T07:13:42.268337139Z ERROR   rw-standalone-compute risingwave_batch::rpc::service::task_service: failed to fire task error=failed to build executor: Expr error: expect user defined
2024-09-06T07:13:42.268198629Z ERROR   rw-standalone-compute risingwave_batch::rpc::service::task_service: failed to fire task error=failed to build executor: Expr error: expect user defined
2024-09-06T07:13:42.268366148Z  INFO          rw-batch-local handle_query{mode="simple query" session_id=4 sql=SELECT *, weighted_avg(exam_id, score_id) OVER (PARTITION BY "student_id" ORDER BY exam_date DESC RANGE BETWEEN '24 hours' PRECEDING AND CURRENT ROW) AS "weighted_avg" FROM exam_scores}:distributed_execute{query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2" epoch=BatchQueryEpoch { epoch: Some(Committed(7102242555625472)) }}:stage{otel.name="Stage 2feee03a-7fa4-45cd-a984-12a93197d3d2-1" query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2" stage_id=1}: risingwave_batch::worker_manager::worker_node_manager: Mask worker node 1 for 30s temporarily
2024-09-06T07:13:42.268257099Z ERROR   rw-standalone-compute risingwave_batch::executor: failed to build executor error=Expr error: expect user defined plan_node=Ok(SortOverWindow(SortOverWindowNode { calls: [WindowFunction { args: [InputRef { index: 1, r#type: Some(INT32) }, InputRef { index: 0, r#type: Some(INT32) }], return_type: Some(DOUBLE), frame: Some(WindowFrame { r#type: Range, start: None, end: None, exclusion: NoOthers, bounds: Some(Range(RangeFrameBounds { start: Some(RangeFrameBound { r#type: Preceding, offset: Some(Datum { body: [1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 96, 215, 29, 20, 0, 0, 0] }) }), end: Some(RangeFrameBound { r#type: CurrentRow, offset: None }), order_data_type: Some(TIMESTAMP), order_type: Some(OrderType { direction: Descending, nulls_are: Largest }), offset_data_type: Some(INTERVAL) })) }), r#type: Some(Aggregate(UserDefined)) }], partition_by: [2], order_by: [ColumnOrder { column_index: 4, order_type: Some(OrderType { direction: Descending, nulls_are: Largest }) }] }))
2024-09-06T07:13:42.268492799Z ERROR          rw-batch-local handle_query{mode="simple query" session_id=4 sql=SELECT *, weighted_avg(exam_id, score_id) OVER (PARTITION BY "student_id" ORDER BY exam_date DESC RANGE BETWEEN '24 hours' PRECEDING AND CURRENT ROW) AS "weighted_avg" FROM exam_scores}:distributed_execute{query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2" epoch=BatchQueryEpoch { epoch: Some(Committed(7102242555625472)) }}:stage{otel.name="Stage 2feee03a-7fa4-45cd-a984-12a93197d3d2-1" query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2" stage_id=1}: risingwave_frontend::scheduler::distributed::stage: Failed to schedule tasks error=gRPC request to batch service failed: Internal error: failed to build executor: Expr error: expect user defined query_id=QueryId { id: "2feee03a-7fa4-45cd-a984-12a93197d3d2" } stage_id=1
2024-09-06T07:13:42.268530419Z ERROR   rw-standalone-compute risingwave_batch::rpc::service::task_service: failed to fire task error=failed to build executor: Expr error: expect user defined
2024-09-06T07:13:42.268171569Z ERROR   rw-standalone-compute risingwave_batch::rpc::service::task_service: failed to fire task error=failed to build executor: Expr error: expect user defined
2024-09-06T07:13:42.268326969Z ERROR   rw-standalone-compute risingwave_batch::rpc::service::task_service: failed to fire task error=failed to build executor: Expr error: expect user defined
2024-09-06T07:13:42.268746879Z ERROR  rw-standalone-frontend handle_query{mode="simple query" session_id=4 sql=SELECT *, weighted_avg(exam_id, score_id) OVER (PARTITION BY "student_id" ORDER BY exam_date DESC RANGE BETWEEN '24 hours' PRECEDING AND CURRENT ROW) AS "weighted_avg" FROM exam_scores}:distributed_execute{query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2" epoch=BatchQueryEpoch { epoch: Some(Committed(7102242555625472)) }}: risingwave_frontend::scheduler::distributed::query: query stage failed error=gRPC request to batch service failed: Internal error: failed to build executor: Expr error: expect user defined query_id=QueryId { id: "2feee03a-7fa4-45cd-a984-12a93197d3d2" } stage_id=1
2024-09-06T07:13:42.268359129Z ERROR   rw-standalone-compute risingwave_batch::rpc::service::task_service: failed to fire task error=failed to build executor: Expr error: expect user defined
2024-09-06T07:13:42.268264189Z ERROR   rw-standalone-compute risingwave_batch::executor: failed to build executor error=Expr error: expect user defined plan_node=Ok(SortOverWindow(SortOverWindowNode { calls: [WindowFunction { args: [InputRef { index: 1, r#type: Some(INT32) }, InputRef { index: 0, r#type: Some(INT32) }], return_type: Some(DOUBLE), frame: Some(WindowFrame { r#type: Range, start: None, end: None, exclusion: NoOthers, bounds: Some(Range(RangeFrameBounds { start: Some(RangeFrameBound { r#type: Preceding, offset: Some(Datum { body: [1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 96, 215, 29, 20, 0, 0, 0] }) }), end: Some(RangeFrameBound { r#type: CurrentRow, offset: None }), order_data_type: Some(TIMESTAMP), order_type: Some(OrderType { direction: Descending, nulls_are: Largest }), offset_data_type: Some(INTERVAL) })) }), r#type: Some(Aggregate(UserDefined)) }], partition_by: [2], order_by: [ColumnOrder { column_index: 4, order_type: Some(OrderType { direction: Descending, nulls_are: Largest }) }] }))
2024-09-06T07:13:42.268936909Z ERROR   rw-standalone-compute risingwave_batch::rpc::service::task_service: failed to fire task error=failed to build executor: Expr error: expect user defined
2024-09-06T07:13:42.268972449Z ERROR  rw-standalone-frontend handle_query{mode="simple query" session_id=4 sql=SELECT *, weighted_avg(exam_id, score_id) OVER (PARTITION BY "student_id" ORDER BY exam_date DESC RANGE BETWEEN '24 hours' PRECEDING AND CURRENT ROW) AS "weighted_avg" FROM exam_scores}: pgwire::pg_protocol: error when process message error=Failed to run the query: Scheduler error: gRPC request to batch service failed: Internal error: failed to build executor: Expr error: expect user defined
2024-09-06T07:13:42.269474109Z ERROR   rw-standalone-compute risingwave_batch::executor: failed to build executor error=Expr error: expect user defined plan_node=Ok(SortOverWindow(SortOverWindowNode { calls: [WindowFunction { args: [InputRef { index: 1, r#type: Some(INT32) }, InputRef { index: 0, r#type: Some(INT32) }], return_type: Some(DOUBLE), frame: Some(WindowFrame { r#type: Range, start: None, end: None, exclusion: NoOthers, bounds: Some(Range(RangeFrameBounds { start: Some(RangeFrameBound { r#type: Preceding, offset: Some(Datum { body: [1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 96, 215, 29, 20, 0, 0, 0] }) }), end: Some(RangeFrameBound { r#type: CurrentRow, offset: None }), order_data_type: Some(TIMESTAMP), order_type: Some(OrderType { direction: Descending, nulls_are: Largest }), offset_data_type: Some(INTERVAL) })) }), r#type: Some(Aggregate(UserDefined)) }], partition_by: [2], order_by: [ColumnOrder { column_index: 4, order_type: Some(OrderType { direction: Descending, nulls_are: Largest }) }] }))
2024-09-06T07:13:42.269818659Z  WARN                rw-batch batch_execute{task_id=12 stage_id=2 query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2"}: risingwave_batch::task::task_execution: The status receiver in FE has closed so the status push is failed error=Failed to send result to channel
2024-09-06T07:13:42.269957359Z  WARN                rw-batch batch_execute{task_id=13 stage_id=2 query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2"}: risingwave_batch::task::task_execution: The status receiver in FE has closed so the status push is failed error=Failed to send result to channel
2024-09-06T07:13:42.270034709Z  WARN                rw-batch batch_execute{task_id=14 stage_id=2 query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2"}: risingwave_batch::task::task_execution: The status receiver in FE has closed so the status push is failed error=Failed to send result to channel
2024-09-06T07:13:42.270013899Z  WARN                rw-batch batch_execute{task_id=8 stage_id=2 query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2"}: risingwave_batch::task::task_execution: The status receiver in FE has closed so the status push is failed error=Failed to send result to channel
2024-09-06T07:13:42.270028249Z  WARN                rw-batch batch_execute{task_id=1 stage_id=2 query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2"}: risingwave_batch::task::task_execution: The status receiver in FE has closed so the status push is failed error=Failed to send result to channel
2024-09-06T07:13:42.270126378Z  WARN                rw-batch batch_execute{task_id=7 stage_id=2 query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2"}: risingwave_batch::task::task_execution: The status receiver in FE has closed so the status push is failed error=Failed to send result to channel
2024-09-06T07:13:42.270111029Z  WARN                rw-batch batch_execute{task_id=15 stage_id=2 query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2"}: risingwave_batch::task::task_execution: The status receiver in FE has closed so the status push is failed error=Failed to send result to channel
2024-09-06T07:13:42.270145759Z  WARN                rw-batch batch_execute{task_id=11 stage_id=2 query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2"}: risingwave_batch::task::task_execution: The status receiver in FE has closed so the status push is failed error=Failed to send result to channel
2024-09-06T07:13:42.269912099Z  WARN                rw-batch batch_execute{task_id=10 stage_id=2 query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2"}: risingwave_batch::task::task_execution: The status receiver in FE has closed so the status push is failed error=Failed to send result to channel
2024-09-06T07:13:42.270207979Z  WARN                rw-batch batch_execute{task_id=6 stage_id=2 query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2"}: risingwave_batch::task::task_execution: The status receiver in FE has closed so the status push is failed error=Failed to send result to channel
2024-09-06T07:13:42.270238779Z  WARN                rw-batch batch_execute{task_id=0 stage_id=2 query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2"}: risingwave_batch::task::task_execution: The status receiver in FE has closed so the status push is failed error=Failed to send result to channel
2024-09-06T07:13:42.270344769Z ERROR   rw-standalone-compute risingwave_batch::rpc::service::task_service: failed to fire task error=failed to build executor: Expr error: expect user defined
2024-09-06T07:13:42.270363869Z  WARN                rw-batch batch_execute{task_id=2 stage_id=2 query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2"}: risingwave_batch::task::task_execution: The status receiver in FE has closed so the status push is failed error=Failed to send result to channel
2024-09-06T07:13:42.270412938Z  WARN                rw-batch batch_execute{task_id=5 stage_id=2 query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2"}: risingwave_batch::task::task_execution: The status receiver in FE has closed so the status push is failed error=Failed to send result to channel
2024-09-06T07:13:42.270455599Z  WARN                rw-batch batch_execute{task_id=9 stage_id=2 query_id="2feee03a-7fa4-45cd-a984-12a93197d3d2"}: risingwave_batch::task::task_execution: The status receiver in FE has closed so the status push is failed error=Failed to send result to channel

To Reproduce

CREATE TABLE exam_scores (
  score_id int,
  exam_id int,
  student_id int,
  score real,
  exam_date timestamp
);

INSERT INTO exam_scores (score_id, exam_id, student_id, score, exam_date)
VALUES
  (1, 101, 1001, 85.5, '2022-01-10'),
  (2, 101, 1002, 92.0, '2022-01-10'),
  (3, 101, 1003, 78.5, '2022-01-10'),
  (4, 102, 1001, 91.2, '2022-02-15'),
  (5, 102, 1003, 88.9, '2022-02-15');
  
create aggregate weighted_avg(value int, weight int) returns float language python as $$
def create_state():
    return (0, 0)

def accumulate(state, value, weight):
    if value is None or weight is None:
        return state
    (s, w) = state
    s += value * weight
    w += weight
    return (s, w)

def retract(state, value, weight):
    if value is None or weight is None:
        return state
    (s, w) = state
    s -= value * weight
    w -= weight
    return (s, w)

def finish(state):
    (sum, weight) = state
    if weight == 0:
        return None
    else:
        return sum / weight
$$;

SELECT *, weighted_avg(exam_id, score_id) OVER (PARTITION BY "student_id" ORDER BY exam_date DESC RANGE BETWEEN '24 hours' PRECEDING AND CURRENT ROW) AS "weighted_avg" FROM exam_scores;
  

Expected behavior

No response

How did you deploy RisingWave?

No response

The version of RisingWave

No response

Additional context

No response