diff --git a/datafusion/functions-window/src/lead_lag.rs b/datafusion/functions-window/src/lead_lag.rs index fea4a1a4aadda..e549f9a68ba1e 100644 --- a/datafusion/functions-window/src/lead_lag.rs +++ b/datafusion/functions-window/src/lead_lag.rs @@ -33,7 +33,7 @@ use datafusion_expr::{ use datafusion_functions_window_common::expr::ExpressionArgs; use datafusion_functions_window_common::field::WindowUDFFieldArgs; use datafusion_functions_window_common::partition::PartitionEvaluatorArgs; -use datafusion_physical_expr::expressions; +use datafusion_physical_expr::expressions::{self, CastExpr}; use datafusion_physical_expr_common::physical_expr::PhysicalExpr; use std::cmp::min; use std::collections::VecDeque; @@ -60,6 +60,12 @@ get_or_init_udwf!( WindowShift::lead ); +#[derive(Debug, Clone)] +pub enum DefaultValue { + Literal(ScalarValue), + Expression, +} + /// Create an expression to represent the `lag` window function /// /// returns value evaluated at the row that is offset rows before the current row within the partition; @@ -178,19 +184,21 @@ static LAG_DOCUMENTATION: LazyLock = LazyLock::new(|| { -- Example usage of the lag window function: SELECT employee_id, salary, - lag(salary, 1, 0) OVER (ORDER BY employee_id) AS prev_salary + lag(salary, 1, 0) OVER (ORDER BY employee_id) AS prev_salary, + lag(salary, 1, salary) OVER (ORDER BY employee_id) AS prev_salary_or_current FROM employees; -+-------------+--------+-------------+ -| employee_id | salary | prev_salary | -+-------------+--------+-------------+ -| 1 | 30000 | 0 | -| 2 | 50000 | 30000 | -| 3 | 70000 | 50000 | -| 4 | 60000 | 70000 | -+-------------+--------+-------------+ ++-------------+--------+-------------+------------------------+ +| employee_id | salary | prev_salary | prev_salary_or_current | ++-------------+--------+-------------+------------------------+ +| 1 | 30000 | 0 | 30000 | +| 2 | 50000 | 30000 | 30000 | +| 3 | 70000 | 50000 | 50000 | +| 4 | 60000 | 70000 | 70000 | ++-------------+--------+-------------+------------------------+ ``` "#) + .build() }); @@ -216,18 +224,19 @@ SELECT employee_id, department, salary, - lead(salary, 1, 0) OVER (PARTITION BY department ORDER BY salary) AS next_salary + lead(salary, 1, 0) OVER (PARTITION BY department ORDER BY salary) AS next_salary, + lead(salary, 1, salary) OVER (PARTITION BY department ORDER BY salary) AS next_salary_or_current FROM employees; -+-------------+-------------+--------+--------------+ -| employee_id | department | salary | next_salary | -+-------------+-------------+--------+--------------+ -| 1 | Sales | 30000 | 50000 | -| 2 | Sales | 50000 | 70000 | -| 3 | Sales | 70000 | 0 | -| 4 | Engineering | 40000 | 60000 | -| 5 | Engineering | 60000 | 0 | -+-------------+-------------+--------+--------------+ ++-------------+-------------+--------+--------------+------------------------+ +| employee_id | department | salary | next_salary | next_salary_or_current | ++-------------+-------------+--------+--------------+------------------------+ +| 1 | Sales | 30000 | 50000 | 50000 | +| 2 | Sales | 50000 | 70000 | 70000 | +| 3 | Sales | 70000 | 0 | 70000 | +| 4 | Engineering | 40000 | 60000 | 60000 | +| 5 | Engineering | 60000 | 0 | 60000 | ++-------------+-------------+--------+--------------+------------------------+ ``` "#) .build() @@ -246,15 +255,51 @@ impl WindowUDFImpl for WindowShift { &self.signature } - /// Handles the case where `NULL` expression is passed as an - /// argument to `lead`/`lag`. The type is refined depending - /// on the default value argument. + /// Handles cases: + /// - where `NULL` expression is passed as an argument to `lead`/`lag`. The type is refined depending + /// on the default value argument. + /// - where input expression contains another expression (PhysicalExpr) + /// in this case, in later evaluate() and evaluate_all() we will have result of applying + /// this PhysicalExpr to the RecordBatch (thus, we can use it as a default value) /// /// For more details see: fn expressions(&self, expr_args: ExpressionArgs) -> Vec> { - parse_expr(expr_args.input_exprs(), expr_args.input_fields()) - .into_iter() - .collect::>() + let input_exprs = expr_args.input_exprs(); + let input_fields = expr_args.input_fields(); + let mut result = Vec::new(); + + let main_expr = + parse_expr(expr_args.input_exprs(), expr_args.input_fields()).unwrap(); + let main_expr_nullable = expr_args + .input_fields() + .first() + .unwrap() + .data_type() + .is_null(); + let main_expr_type = input_fields[0].data_type(); + result.push(main_expr); + + // Pushing the expression (not a literal value) to the result, so it would be executed + // If our main (first argument) expression is nullable => no type casting involved + if main_expr_nullable { + if input_exprs.len() >= 3 { + result.push(Arc::clone(&input_exprs[2])); + } + } else { + if input_exprs.len() >= 3 { + let default_expr_type = input_fields[2].data_type(); + if default_expr_type == main_expr_type { + result.push(Arc::clone(&input_exprs[2])); + } else { + result.push(Arc::new(CastExpr::new( + Arc::clone(&input_exprs[2]), + main_expr_type.to_owned(), + None, + ))) + } + } + } + result } fn partition_evaluator( @@ -393,20 +438,15 @@ fn parse_expr_field(input_fields: &[FieldRef]) -> Result { fn parse_default_value( input_exprs: &[Arc], input_types: &[FieldRef], -) -> Result { +) -> Result { let expr_field = parse_expr_field(input_types)?; - let unparsed = get_scalar_value_from_args(input_exprs, 2)?; - - unparsed - .filter(|v| !v.data_type().is_null()) - .map(|v| v.cast_to(expr_field.data_type())) - .unwrap_or_else(|| ScalarValue::try_from(expr_field.data_type())) + get_default_value_from_args(input_exprs, 2, &expr_field) } #[derive(Debug)] struct WindowShiftEvaluator { shift_offset: i64, - default_value: ScalarValue, + default_value: DefaultValue, ignore_nulls: bool, // VecDeque contains offset values that between non-null entries non_null_offsets: VecDeque, @@ -566,6 +606,120 @@ fn shift_with_default_value( } } +fn shift_with_array_default( + array: &ArrayRef, + offset: i64, + default_values: &ArrayRef, +) -> Result { + use datafusion_common::arrow::compute::concat; + + let value_len = array.len() as i64; + if offset == 0 { + return Ok(Arc::clone(array)); + } + if offset == i64::MIN || offset.abs() >= value_len { + return Ok(Arc::clone(default_values)); + } + + let slice_offset = (-offset).clamp(0, value_len) as usize; + let length = array.len() - offset.unsigned_abs() as usize; + let slice = array.slice(slice_offset, length); + + let defaults_slice = if offset > 0 { + // Lag: defaults go at the beginning + default_values.slice(0, offset.unsigned_abs() as usize) + } else { + // Lead: defaults go at the end + let start = default_values.len() - offset.unsigned_abs() as usize; + default_values.slice(start, offset.unsigned_abs() as usize) + }; + + if offset > 0 { + concat(&[defaults_slice.as_ref(), slice.as_ref()]) + } else { + concat(&[slice.as_ref(), defaults_slice.as_ref()]) + } + .map_err(|e| arrow_datafusion_err!(e)) +} + +fn evaluate_all_with_ignore_null_and_array_default( + array: &ArrayRef, + offset: i64, + default_values: &ArrayRef, + is_lag: bool, +) -> Result { + // Arrays without NULLs do not necessarily have a null bitmap. + // Note: https://arrow.apache.org/docs/format/Columnar.html#validity-bitmaps + let Some(nulls) = array.nulls() else { + return shift_with_array_default(array, offset, default_values); + }; + + let valid_indices: Vec = nulls.valid_indices().collect::>(); + let direction = !is_lag; + let results: Result> = (0..array.len()) + .map(|id| { + let result_index = match valid_indices.binary_search(&id) { + Ok(pos) => if direction { + pos.checked_add(offset as usize) + } else { + pos.checked_sub(offset.unsigned_abs() as usize) + } + .and_then(|new_pos| { + if new_pos < valid_indices.len() { + Some(valid_indices[new_pos]) + } else { + None + } + }), + Err(pos) => if direction { + pos.checked_add(offset as usize) + } else if pos > 0 { + pos.checked_sub(offset.unsigned_abs() as usize) + } else { + None + } + .and_then(|new_pos| { + if new_pos < valid_indices.len() { + Some(valid_indices[new_pos]) + } else { + None + } + }), + }; + match result_index { + Some(index) => ScalarValue::try_from_array(array, index), + None => ScalarValue::try_from_array(&default_values, id), + } + }) + .collect(); + ScalarValue::iter_to_array(results?) +} + +pub(crate) fn get_default_value_from_args( + args: &[Arc], + index: usize, + field: &Arc, +) -> Result { + match args.get(index) { + Some(expr) => { + if let Some(literal) = expr.downcast_ref::() { + let scalar = literal.value().clone(); + let scalar = if !scalar.data_type().is_null() { + scalar.cast_to(field.data_type()) + } else { + ScalarValue::try_from(field.data_type()) + }?; + Ok(DefaultValue::Literal(scalar)) + } else { + Ok(DefaultValue::Expression) + } + } + None => Ok(DefaultValue::Literal(ScalarValue::try_from( + field.data_type(), + )?)), + } +} + impl PartitionEvaluator for WindowShiftEvaluator { fn get_range(&self, idx: usize, n_rows: usize) -> Result> { let offset = offset_magnitude(self.shift_offset); @@ -712,7 +866,29 @@ impl PartitionEvaluator for WindowShiftEvaluator { if !(idx.is_none() || (self.ignore_nulls && array.is_null(idx.unwrap()))) { ScalarValue::try_from_array(array, idx.unwrap()) } else { - Ok(self.default_value.clone()) + match &self.default_value { + DefaultValue::Literal(scalar) => Ok(scalar.clone()), + DefaultValue::Expression => { + let current_row = if self.is_lag() { + range.end.saturating_sub(1) + } else { + range.start + }; + + values + .get(1) + .map(|defaults| { + let scalar = + ScalarValue::try_from_array(defaults, current_row)?; + if scalar.data_type() != *array.data_type() { + scalar.cast_to(array.data_type()) + } else { + Ok(scalar) + } + }) + .unwrap_or_else(|| ScalarValue::try_from(array.data_type())) + } + } } } @@ -721,17 +897,38 @@ impl PartitionEvaluator for WindowShiftEvaluator { values: &[ArrayRef], _num_rows: usize, ) -> Result { - // LEAD, LAG window functions take single column, values will have size 1 + // LEAD, LAG window functions take single column, values will have size: + // '1' - when default_value is a ScalarValue (or we simply did not specify it) + // '2' - when default_value is a PhysicalExpr let value = &values[0]; - if !self.ignore_nulls { - shift_with_default_value(value, self.shift_offset, &self.default_value) - } else { - evaluate_all_with_ignore_null( - value, - self.shift_offset, - &self.default_value, - self.is_lag(), - ) + match &self.default_value { + DefaultValue::Literal(scalar) => { + if !self.ignore_nulls { + shift_with_default_value(value, self.shift_offset, &scalar.clone()) + } else { + evaluate_all_with_ignore_null( + value, + self.shift_offset, + &scalar.clone(), + self.is_lag(), + ) + } + } + DefaultValue::Expression => { + let default_array = values.get(1).cloned().unwrap_or_else(|| { + Arc::new(arrow::array::NullArray::new(value.len())) + }); + if !self.ignore_nulls { + shift_with_array_default(value, self.shift_offset, &default_array) + } else { + evaluate_all_with_ignore_null_and_array_default( + value, + self.shift_offset, + &default_array, + self.is_lag(), + ) + } + } } } @@ -769,7 +966,7 @@ mod tests { // LAG(2) let lag_fn = WindowShiftEvaluator { shift_offset: 2, - default_value: ScalarValue::Null, + default_value: DefaultValue::Literal(ScalarValue::Null), ignore_nulls: false, non_null_offsets: Default::default(), }; @@ -779,7 +976,7 @@ mod tests { // LAG(2 ignore nulls) let lag_fn = WindowShiftEvaluator { shift_offset: 2, - default_value: ScalarValue::Null, + default_value: DefaultValue::Literal(ScalarValue::Null), ignore_nulls: true, // models data received [, , , NULL, , NULL, , ...] non_null_offsets: vec![2, 2].into(), // [1, 1, 2, 2] actually, just last 2 is used @@ -789,7 +986,7 @@ mod tests { // LEAD(2) let lead_fn = WindowShiftEvaluator { shift_offset: -2, - default_value: ScalarValue::Null, + default_value: DefaultValue::Literal(ScalarValue::Null), ignore_nulls: false, non_null_offsets: Default::default(), }; @@ -799,7 +996,7 @@ mod tests { // LEAD(2 ignore nulls) let lead_fn = WindowShiftEvaluator { shift_offset: -2, - default_value: ScalarValue::Null, + default_value: DefaultValue::Literal(ScalarValue::Null), ignore_nulls: true, // models data received [..., , NULL, , NULL, , ..] non_null_offsets: vec![2, 2].into(), diff --git a/datafusion/sqllogictest/test_files/window.slt b/datafusion/sqllogictest/test_files/window.slt index cbbd9b74dfc00..c56257d3ccfbe 100644 --- a/datafusion/sqllogictest/test_files/window.slt +++ b/datafusion/sqllogictest/test_files/window.slt @@ -4169,30 +4169,328 @@ select arrow_typeof(nth_value(a, 1) over ()) from (select 1 a) ---- Int64 +# === Basic LEAD/LAG with NULL default (2 rows) === + # test LEAD window function works NULL as default value query I -select lead(a, 1, null) over (order by a) from (select 1 a union all select 2 a) +WITH t(a) AS (VALUES (1), (2)) +SELECT lead(a, 1, null) OVER (ORDER BY a) FROM t ---- 2 NULL # test LAG window function works NULL as default value query I -select lag(a, 1, null) over (order by a) from (select 1 a union all select 2 a) +WITH t(a) AS (VALUES (1), (2)) +SELECT lag(a, 1, null) OVER (ORDER BY a) FROM t ---- NULL 1 + +# === Basic LEAD/LAG with column default (3 rows with a, b) === + +# test LAG with another column as default (first row falls back to b) +query II +WITH t(a, b) AS (VALUES (1, 10), (2, 20), (3, 30)) +SELECT a, lag(a, 1, b) OVER (ORDER BY a) FROM t +---- +1 10 +2 1 +3 2 + +# test LEAD with another column as default (last row falls back to b) +query II +WITH t(a, b) AS (VALUES (1, 10), (2, 20), (3, 30)) +SELECT a, lead(a, 1, b) OVER (ORDER BY a) FROM t +---- +1 2 +2 3 +3 30 + + +# === Offset 2 with column default (4 rows with a, b) === + +# test LAG with offset 2 and column default (first 2 rows fall back to b) +query II +WITH t(a, b) AS (VALUES (1, 10), (2, 20), (3, 30), (4, 40)) +SELECT a, lag(a, 2, b) OVER (ORDER BY a) FROM t +---- +1 10 +2 20 +3 1 +4 2 + +# test LEAD with offset 2 and column default (last 2 rows fall back to b) +query II +WITH t(a, b) AS (VALUES (1, 10), (2, 20), (3, 30), (4, 40)) +SELECT a, lead(a, 2, b) OVER (ORDER BY a) FROM t +---- +1 3 +2 4 +3 30 +4 40 + + +# === Expression defaults (4 rows with a, b) === + +# test LAG with column expression default +query II +WITH t(a, b) AS (VALUES (1, 10), (2, 20), (3, 30), (4, 40)) +SELECT a, lag(a, 1, b / 2) OVER (ORDER BY a) FROM t +---- +1 5 +2 1 +3 2 +4 3 + +# test LAG with mixed-column expression default +query II +WITH t(a, b, c) AS ( + VALUES (1, 10, 100), (2, 20, 200), (3, 30, 300), (4, 40, 400) +) +SELECT a, lag(a, 1, b * c) OVER (ORDER BY a) FROM t +---- +1 1000 +2 1 +3 2 +4 3 + + +# === Simple expression defaults on single column (2-4 rows) === + +# test LAG with a * 2 as default +query II +WITH t(a) AS (VALUES (1), (2)) +SELECT a, lag(a, 1, a * 2) OVER (ORDER BY a) FROM t +---- +1 2 +2 1 + +# test LAG with offset 2 and a * 3 as default +query II +WITH t(a) AS (VALUES (1), (2), (3)) +SELECT a, lag(a, 2, a * 3) OVER (ORDER BY a) FROM t +---- +1 3 +2 6 +3 1 + +# test LEAD with a * 2 as default +query II +WITH t(a) AS (VALUES (1), (2), (3)) +SELECT a, lead(a, 1, a * 2) OVER (ORDER BY a) FROM t +---- +1 2 +2 3 +3 6 + +# test LEAD with offset 2 and a * 3 as default +query II +WITH t(a) AS (VALUES (1), (2), (3), (4)) +SELECT a, lead(a, 2, a * 3) OVER (ORDER BY a) FROM t +---- +1 3 +2 4 +3 9 +4 12 + + +# === Combined LAG and LEAD (3 rows) === + +# test Both lag and lead in same query with expression defaults +query III +WITH t(a) AS (VALUES (1), (2), (3)) +SELECT a, + lag(a, 1, a * 2) OVER (ORDER BY a), + lead(a, 1, a * 3) OVER (ORDER BY a) +FROM t +---- +1 2 2 +2 1 3 +3 2 9 + + +# === Multi-column expression defaults (3 rows with a, b, c) === + +# test LAG default is sum of two columns +query II +WITH t(a, b, c) AS (VALUES (1, 10, 100), (2, 20, 200), (3, 30, 300)) +SELECT a, lag(a, 1, b + c) OVER (ORDER BY a) FROM t +---- +1 110 +2 1 +3 2 + +# test LEAD default is product of two columns +query II +WITH t(a, b, c) AS (VALUES (1, 2, 3), (2, 4, 5), (3, 6, 7)) +SELECT a, lead(a, 1, b * c) OVER (ORDER BY a) FROM t +---- +1 2 +2 3 +3 42 + + +# === PARTITION BY with column defaults === + +# test LAG with PARTITION BY and column default +query III +WITH t(dept, salary) AS ( + VALUES (1, 1000), (1, 2000), (2, 3000), (2, 4000) +) +SELECT dept, salary, lag(salary, 1, salary * 2) OVER (PARTITION BY dept ORDER BY salary) +FROM t +ORDER BY dept, salary +---- +1 1000 2000 +1 2000 1000 +2 3000 6000 +2 4000 3000 + +# test LEAD with PARTITION BY and column default +query III +WITH t(dept, salary) AS ( + VALUES (1, 1000), (1, 2000), (2, 3000), (2, 4000) +) +SELECT dept, salary, lead(salary, 1, salary * 2) OVER (PARTITION BY dept ORDER BY salary) +FROM t +ORDER BY dept, salary +---- +1 1000 2000 +1 2000 4000 +2 3000 4000 +2 4000 8000 + + +# === NULL handling === + +# test LEAD with NULLs in main column and column default +query II +WITH t(a, b) AS (VALUES (1, 1), (NULL, 2), (3, 3)) +SELECT a, lead(a, 1, b) OVER (ORDER BY b) FROM t +---- +1 NULL +NULL 3 +3 3 + + +# === Single row tests === + +# test LAG on single row — always returns default expression value +query II +WITH t(a) AS (VALUES (5)) +SELECT a, lag(a, 1, a * 99) OVER (ORDER BY a) FROM t +---- +5 495 + +# test LEAD on single row — always returns default expression value +query II +WITH t(a) AS (VALUES (5)) +SELECT a, lead(a, 1, a * 99) OVER (ORDER BY a) FROM t +---- +5 495 + + +# === Row-specific defaults (4 rows) === + +# test each row's default is its own value * 10; only first/last rows use it +query II +WITH t(a) AS (VALUES (1), (2), (3), (4)) +SELECT a, lag(a, 1, a * 10) OVER (ORDER BY a) FROM t +---- +1 10 +2 1 +3 2 +4 3 + +# test lead with rows'default value as its own * 10 +query II +WITH t(a) AS (VALUES (1), (2), (3), (4)) +SELECT a, lead(a, 1, a * 10) OVER (ORDER BY a) FROM t +---- +1 2 +2 3 +3 4 +4 40 + + +# === Literal vs expression defaults in same query (3 rows) === + +# test literal default and expression default in same query +query III +WITH t(a) AS (VALUES (1), (2), (3)) +SELECT a, + lag(a, 1, 0) OVER (ORDER BY a) AS lag_literal, + lag(a, 1, a * 2) OVER (ORDER BY a) AS lag_expr +FROM t +---- +1 0 2 +2 1 1 +3 2 2 + +query III +WITH t(a) AS (VALUES (1), (2), (3)) +SELECT a, + lead(a, 1, 0) OVER (ORDER BY a) AS lead_literal, + lead(a, 1, a * 2) OVER (ORDER BY a) AS lead_expr +FROM t +---- +1 2 2 +2 3 3 +3 0 6 + + +# === Large offset tests (5 rows) === + +# test LAG offset 3: first 3 rows use default +query II +WITH t(a) AS (VALUES (1), (2), (3), (4), (5)) +SELECT a, lag(a, 3, a * 100) OVER (ORDER BY a) FROM t +---- +1 100 +2 200 +3 300 +4 1 +5 2 + +# test LEAD offset 3: last 3 rows use default +query II +WITH t(a) AS (VALUES (1), (2), (3), (4), (5)) +SELECT a, lead(a, 3, a * 100) OVER (ORDER BY a) FROM t +---- +1 4 +2 5 +3 300 +4 400 +5 500 + +# test LAG with very large offset: all rows use default +query II +WITH t(a) AS (VALUES (1), (2), (3), (4), (5)) +SELECT a, lead(a, 100, a * 10) OVER (ORDER BY a) FROM t +---- +1 10 +2 20 +3 30 +4 40 +5 50 + + +# === String defaults === + # test LEAD window function with string default value query T -select lead(a, 1, 'default') over (order by a) from (select '1' a union all select '2' a) +WITH t(a) AS (VALUES ('1'), ('2')) +SELECT lead(a, 1, 'default') OVER (ORDER BY a) FROM t ---- 2 default # test LAG window function with string default value query T -select lag(a, 1, 'default') over (order by a) from (select '1' a union all select '2' a) +WITH t(a) AS (VALUES ('1'), ('2')) +SELECT lag(a, 1, 'default') OVER (ORDER BY a) FROM t ---- default 1 @@ -6841,4 +7139,4 @@ statement ok DROP TABLE issue_20194_t1; statement ok -DROP TABLE issue_20194_t2; +DROP TABLE issue_20194_t2; \ No newline at end of file diff --git a/docs/source/user-guide/sql/window_functions.md b/docs/source/user-guide/sql/window_functions.md index 2c8050ce1f9ca..9abadcf9cd8e2 100644 --- a/docs/source/user-guide/sql/window_functions.md +++ b/docs/source/user-guide/sql/window_functions.md @@ -403,17 +403,18 @@ lag(expression, offset, default) -- Example usage of the lag window function: SELECT employee_id, salary, - lag(salary, 1, 0) OVER (ORDER BY employee_id) AS prev_salary + lag(salary, 1, 0) OVER (ORDER BY employee_id) AS prev_salary, + lag(salary, 1, salary) OVER (ORDER BY employee_id) AS prev_salary_or_current FROM employees; -+-------------+--------+-------------+ -| employee_id | salary | prev_salary | -+-------------+--------+-------------+ -| 1 | 30000 | 0 | -| 2 | 50000 | 30000 | -| 3 | 70000 | 50000 | -| 4 | 60000 | 70000 | -+-------------+--------+-------------+ ++-------------+--------+-------------+------------------------+ +| employee_id | salary | prev_salary | prev_salary_or_current | ++-------------+--------+-------------+------------------------+ +| 1 | 30000 | 0 | 30000 | +| 2 | 50000 | 30000 | 30000 | +| 3 | 70000 | 50000 | 50000 | +| 4 | 60000 | 70000 | 70000 | ++-------------+--------+-------------+------------------------+ ``` ### `last_value` @@ -471,18 +472,19 @@ SELECT employee_id, department, salary, - lead(salary, 1, 0) OVER (PARTITION BY department ORDER BY salary) AS next_salary + lead(salary, 1, 0) OVER (PARTITION BY department ORDER BY salary) AS next_salary, + lead(salary, 1, salary) OVER (PARTITION BY department ORDER BY salary) AS next_salary_or_current FROM employees; -+-------------+-------------+--------+--------------+ -| employee_id | department | salary | next_salary | -+-------------+-------------+--------+--------------+ -| 1 | Sales | 30000 | 50000 | -| 2 | Sales | 50000 | 70000 | -| 3 | Sales | 70000 | 0 | -| 4 | Engineering | 40000 | 60000 | -| 5 | Engineering | 60000 | 0 | -+-------------+-------------+--------+--------------+ ++-------------+-------------+--------+--------------+------------------------+ +| employee_id | department | salary | next_salary | next_salary_or_current | ++-------------+-------------+--------+--------------+------------------------+ +| 1 | Sales | 30000 | 50000 | 50000 | +| 2 | Sales | 50000 | 70000 | 70000 | +| 3 | Sales | 70000 | 0 | 70000 | +| 4 | Engineering | 40000 | 60000 | 60000 | +| 5 | Engineering | 60000 | 0 | 60000 | ++-------------+-------------+--------+--------------+------------------------+ ``` ### `nth_value`