feat(expr): support IGNORE NULLS for first_value/last_value

e2e_test/over_window/generated/batch/ignore_nulls/mod.slt.part

@@ -0,0 +1,51 @@
+# This file is generated by `gen.py`. Do not edit it manually!
+
+# Test `IGNORE NULLS` in window function argument lists.
+
+statement ok
+create table t (ts timestamptz, val int);
+
+statement ok
+create view v as
+select
+    *,
+    first_value(val) over (partition by 1::int order by ts) as first,
+    first_value(val ignore nulls) over (partition by 1::int order by ts) as first_non_null,
+    first_value(val ignore nulls) over (partition by 1::int order by ts rows between 1 preceding and 1 following) as first_non_null_2,
+    last_value(val) over (partition by 1::int order by ts) as last,
+    last_value(val ignore nulls) over (partition by 1::int order by ts) as last_non_null,
+    last_value(val ignore nulls) over (partition by 1::int order by ts rows between 1 preceding and 1 following) as last_non_null_2
+from t;
+
+statement ok
+insert into t values
+  ('2024-12-18T12:00:42Z', 3)
+, ('2024-12-18T12:01:00Z', 2)
+, ('2024-12-18T12:01:15Z', null);
+
+query iiii
+select first, first_non_null, first_non_null_2, last, last_non_null, last_non_null_2 from v order by ts;
+----
+3    3    3    3    3    3
+3    3    3    2    2    2
+3    3    3    null 2    2
+
+statement ok
+insert into t values
+  ('2024-12-18T12:03:00Z', null)
+, ('2024-12-18T12:12:00Z', 7);
+
+query iiii
+select first, first_non_null, first_non_null_2, last, last_non_null, last_non_null_2 from v order by ts;
+----
+3    3    3    3    3    2
+3    3    3    2    2    2
+3    3    2    null 2    2
+3    3    7    null 2    7
+3    3    7    7    7    7
+
+statement ok
+drop view v;
+
+statement ok
+drop table t;

e2e_test/over_window/generated/streaming/ignore_nulls/mod.slt.part

@@ -0,0 +1,51 @@
+# This file is generated by `gen.py`. Do not edit it manually!
+
+# Test `IGNORE NULLS` in window function argument lists.
+
+statement ok
+create table t (ts timestamptz, val int);
+
+statement ok
+create materialized view v as
+select
+    *,
+    first_value(val) over (partition by 1::int order by ts) as first,
+    first_value(val ignore nulls) over (partition by 1::int order by ts) as first_non_null,
+    first_value(val ignore nulls) over (partition by 1::int order by ts rows between 1 preceding and 1 following) as first_non_null_2,
+    last_value(val) over (partition by 1::int order by ts) as last,
+    last_value(val ignore nulls) over (partition by 1::int order by ts) as last_non_null,
+    last_value(val ignore nulls) over (partition by 1::int order by ts rows between 1 preceding and 1 following) as last_non_null_2
+from t;
+
+statement ok
+insert into t values
+  ('2024-12-18T12:00:42Z', 3)
+, ('2024-12-18T12:01:00Z', 2)
+, ('2024-12-18T12:01:15Z', null);
+
+query iiii
+select first, first_non_null, first_non_null_2, last, last_non_null, last_non_null_2 from v order by ts;
+----
+3    3    3    3    3    3
+3    3    3    2    2    2
+3    3    3    null 2    2
+
+statement ok
+insert into t values
+  ('2024-12-18T12:03:00Z', null)
+, ('2024-12-18T12:12:00Z', 7);
+
+query iiii
+select first, first_non_null, first_non_null_2, last, last_non_null, last_non_null_2 from v order by ts;
+----
+3    3    3    3    3    2
+3    3    3    2    2    2
+3    3    2    null 2    2
+3    3    7    null 2    7
+3    3    7    7    7    7
+
+statement ok
+drop materialized view v;
+
+statement ok
+drop table t;

e2e_test/over_window/templates/ignore_nulls/mod.slt.part

@@ -0,0 +1,49 @@
+# Test `IGNORE NULLS` in window function argument lists.
+
+statement ok
+create table t (ts timestamptz, val int);
+
+statement ok
+create $view_type v as
+select
+    *,
+    first_value(val) over (partition by 1::int order by ts) as first,
+    first_value(val ignore nulls) over (partition by 1::int order by ts) as first_non_null,
+    first_value(val ignore nulls) over (partition by 1::int order by ts rows between 1 preceding and 1 following) as first_non_null_2,
+    last_value(val) over (partition by 1::int order by ts) as last,
+    last_value(val ignore nulls) over (partition by 1::int order by ts) as last_non_null,
+    last_value(val ignore nulls) over (partition by 1::int order by ts rows between 1 preceding and 1 following) as last_non_null_2
+from t;
+
+statement ok
+insert into t values
+  ('2024-12-18T12:00:42Z', 3)
+, ('2024-12-18T12:01:00Z', 2)
+, ('2024-12-18T12:01:15Z', null);
+
+query iiii
+select first, first_non_null, first_non_null_2, last, last_non_null, last_non_null_2 from v order by ts;
+----
+3    3    3    3    3    3
+3    3    3    2    2    2
+3    3    3    null 2    2
+
+statement ok
+insert into t values
+  ('2024-12-18T12:03:00Z', null)
+, ('2024-12-18T12:12:00Z', 7);
+
+query iiii
+select first, first_non_null, first_non_null_2, last, last_non_null, last_non_null_2 from v order by ts;
+----
+3    3    3    3    3    2
+3    3    3    2    2    2
+3    3    2    null 2    2
+3    3    7    null 2    7
+3    3    7    7    7    7
+
+statement ok
+drop $view_type v;
+
+statement ok
+drop table t;

proto/expr.proto

@@ -590,6 +590,7 @@ message WindowFunction {
   repeated InputRef args = 3;
   data.DataType return_type = 4;
   WindowFrame frame = 5;
+  bool ignore_nulls = 6;
 }
 
 // Note: due to historic reasons, UserDefinedFunction is a oneof variant parallel to FunctionCall,

src/expr/core/src/window_function/call.rs

@@ -31,17 +31,19 @@ use crate::Result;
 #[derive(Debug, Clone)]
 pub struct WindowFuncCall {
     pub kind: WindowFuncKind,
-    pub args: AggArgs,
     pub return_type: DataType,
+    pub args: AggArgs,
+    pub ignore_nulls: bool,
     pub frame: Frame,
 }
 
 impl WindowFuncCall {
     pub fn from_protobuf(call: &PbWindowFunction) -> Result<Self> {
         let call = WindowFuncCall {
             kind: WindowFuncKind::from_protobuf(call.get_type()?)?,
-            args: AggArgs::from_protobuf(call.get_args())?,
             return_type: DataType::from(call.get_return_type()?),
+            args: AggArgs::from_protobuf(call.get_args())?,
+            ignore_nulls: call.get_ignore_nulls(),
             frame: Frame::from_protobuf(call.get_frame()?)?,
         };
         Ok(call)

src/expr/impl/src/window_function/aggregate.rs

@@ -43,6 +43,7 @@ where
 {
     agg_impl: AggImpl,
     arg_data_types: Vec<DataType>,
+    ignore_nulls: bool,
     buffer: WindowBuffer<W>,
     buffer_heap_size: KvSize,
 }
@@ -97,6 +98,7 @@ pub(super) fn new(call: &WindowFuncCall) -> Result<BoxedWindowState> {
         FrameBounds::Rows(frame_bounds) => Box::new(AggregateState {
             agg_impl,
             arg_data_types,
+            ignore_nulls: call.ignore_nulls,
             buffer: WindowBuffer::<RowsWindow<StateKey, StateValue>>::new(
                 RowsWindow::new(frame_bounds.clone()),
                 call.frame.exclusion,
@@ -107,6 +109,7 @@ pub(super) fn new(call: &WindowFuncCall) -> Result<BoxedWindowState> {
         FrameBounds::Range(frame_bounds) => Box::new(AggregateState {
             agg_impl,
             arg_data_types,
+            ignore_nulls: call.ignore_nulls,
             buffer: WindowBuffer::<RangeWindow<StateValue>>::new(
                 RangeWindow::new(frame_bounds.clone()),
                 call.frame.exclusion,
@@ -117,6 +120,7 @@ pub(super) fn new(call: &WindowFuncCall) -> Result<BoxedWindowState> {
         FrameBounds::Session(frame_bounds) => Box::new(AggregateState {
             agg_impl,
             arg_data_types,
+            ignore_nulls: call.ignore_nulls,
             buffer: WindowBuffer::<SessionWindow<StateValue>>::new(
                 SessionWindow::new(frame_bounds.clone()),
                 call.frame.exclusion,
@@ -194,14 +198,31 @@ where
                 wrapper.update(state, self.buffer.consume_curr_window_values_delta())
             }
             AggImpl::Shortcut(shortcut) => match shortcut {
-                Shortcut::FirstValue => Ok(self
-                    .buffer
-                    .curr_window_first_value()
-                    .and_then(|args| args[0].clone())),
-                Shortcut::LastValue => Ok(self
-                    .buffer
-                    .curr_window_last_value()
-                    .and_then(|args| args[0].clone())),
+                Shortcut::FirstValue => Ok(if !self.ignore_nulls {
+                    // no `IGNORE NULLS`
+                    self.buffer
+                        .curr_window_values()
+                        .next()
+                        .and_then(|args| args[0].clone())
+                } else {
+                    // filter out NULLs
+                    self.buffer
+                        .curr_window_values()
+                        .find(|args| args[0].is_some())
+                        .and_then(|args| args[0].clone())
+                }),
+                Shortcut::LastValue => Ok(if !self.ignore_nulls {
+                    self.buffer
+                        .curr_window_values()
+                        .next_back()
+                        .and_then(|args| args[0].clone())
+                } else {
+                    self.buffer
+                        .curr_window_values()
+                        .rev()
+                        .find(|args| args[0].is_some())
+                        .and_then(|args| args[0].clone())
+                }),
             },
         }?;
         let evict_hint = self.slide_inner();

src/expr/impl/src/window_function/buffer.rs

@@ -121,7 +121,7 @@ impl<W: WindowImpl> WindowBuffer<W> {
     }
 
     /// Iterate over values in the current window.
-    pub fn curr_window_values(&self) -> impl Iterator<Item = &W::Value> {
+    pub fn curr_window_values(&self) -> impl DoubleEndedIterator<Item = &W::Value> {
         assert!(self.left_idx <= self.right_excl_idx);
         assert!(self.right_excl_idx <= self.buffer.len());
 
@@ -132,27 +132,6 @@ impl<W: WindowImpl> WindowBuffer<W> {
             .map(|Entry { value, .. }| value)
     }
 
-    /// Get the first value in the current window. Time complexity is O(1).
-    pub fn curr_window_first_value(&self) -> Option<&W::Value> {
-        self.curr_window_values().next()
-    }
-
-    /// Get the last value in the current window. Time complexity is O(1).
-    pub fn curr_window_last_value(&self) -> Option<&W::Value> {
-        let (left, right) = self.curr_window_ranges();
-        if !right.is_empty() {
-            self.buffer
-                .get(right.end - 1)
-                .map(|Entry { value, .. }| value)
-        } else if !left.is_empty() {
-            self.buffer
-                .get(left.end - 1)
-                .map(|Entry { value, .. }| value)
-        } else {
-            None
-        }
-    }
-
     /// Consume the delta of values comparing the current window to the previous window.
     /// The delta is not guaranteed to be sorted, especially when frame exclusion is not `NoOthers`.
     pub fn consume_curr_window_values_delta(

src/expr/impl/src/window_function/rank.rs

@@ -198,8 +198,9 @@ mod tests {
     fn test_rank_state_bad_use() {
         let call = WindowFuncCall {
             kind: WindowFuncKind::RowNumber,
-            args: AggArgs::default(),
             return_type: DataType::Int64,
+            args: AggArgs::default(),
+            ignore_nulls: false,
             frame: Frame::rows(
                 FrameBound::UnboundedPreceding,
                 FrameBound::UnboundedFollowing,
@@ -215,8 +216,9 @@ mod tests {
     fn test_row_number_state() {
         let call = WindowFuncCall {
             kind: WindowFuncKind::RowNumber,
-            args: AggArgs::default(),
             return_type: DataType::Int64,
+            args: AggArgs::default(),
+            ignore_nulls: false,
             frame: Frame::rows(
                 FrameBound::UnboundedPreceding,
                 FrameBound::UnboundedFollowing,
@@ -257,8 +259,9 @@ mod tests {
     fn test_rank_state() {
         let call = WindowFuncCall {
             kind: WindowFuncKind::Rank,
-            args: AggArgs::default(),
             return_type: DataType::Int64,
+            args: AggArgs::default(),
+            ignore_nulls: false,
             frame: Frame::rows(
                 FrameBound::UnboundedPreceding,
                 FrameBound::UnboundedFollowing,
@@ -298,8 +301,9 @@ mod tests {
     fn test_dense_rank_state() {
         let call = WindowFuncCall {
             kind: WindowFuncKind::DenseRank,
-            args: AggArgs::default(),
             return_type: DataType::Int64,
+            args: AggArgs::default(),
+            ignore_nulls: false,
             frame: Frame::rows(
                 FrameBound::UnboundedPreceding,
                 FrameBound::UnboundedFollowing,

src/frontend/planner_test/tests/testdata/input/over_window_function.yaml

@@ -641,3 +641,51 @@
     from t;
   expected_outputs:
     - binder_error
+
+# IGNORE NULLS
+- sql: |
+    create table t (ts timestamptz, val int);
+    select
+      first_value(val ignore nulls) over (partition by 1::int order by ts)
+    from t;
+  expected_outputs:
+    - logical_plan
+    - stream_plan
+    - batch_plan
+- sql: |
+    create table t (ts timestamptz, val int);
+    select
+      last_value(val ignore nulls) over (partition by 1::int order by ts rows between 1 preceding and 1 following)
+    from t;
+  expected_outputs:
+    - logical_plan
+    - stream_plan
+    - batch_plan
+- sql: |
+    create table t (ts timestamptz, val int);
+    select
+      first_value(val) filter (where val is not null) over (partition by 1::int order by ts) -- not supported yet
+    from t;
+  expected_outputs:
+    - binder_error
+- sql: |
+    create table t (ts timestamptz, val int);
+    select
+      last_value(val) filter (where val is not null) over (partition by 1::int order by ts) -- not supported yet
+    from t;
+  expected_outputs:
+    - binder_error
+- sql: |
+    create table t (ts timestamptz, val int);
+    select
+      lag(val ignore nulls) over (partition by 1::int order by ts) -- not supported yet
+    from t;
+  expected_outputs:
+    - binder_error
+- sql: |
+    create table t (ts timestamptz, val int);
+    select
+      lead(val ignore nulls) over (partition by 1::int order by ts) -- not supported yet
+    from t;
+  expected_outputs:
+    - binder_error