Fix LEAD/LAG window functions when default value null (#8989)

datafusion/physical-expr/src/window/lead_lag.rs

@@ -23,8 +23,9 @@ use crate::PhysicalExpr;
 use arrow::array::ArrayRef;
 use arrow::compute::cast;
 use arrow::datatypes::{DataType, Field};
-use datafusion_common::{arrow_datafusion_err, ScalarValue};
-use datafusion_common::{internal_err, DataFusionError, Result};
+use datafusion_common::{
+    arrow_datafusion_err, exec_err, DataFusionError, Result, ScalarValue,
+};
 use datafusion_expr::PartitionEvaluator;
 use std::any::Any;
 use std::cmp::min;
@@ -236,14 +237,16 @@ fn get_default_value(
     default_value: Option<&ScalarValue>,
     dtype: &DataType,
 ) -> Result<ScalarValue> {
-    if let Some(value) = default_value {
-        if let ScalarValue::Int64(Some(val)) = value {
-            ScalarValue::try_from_string(val.to_string(), dtype)
-        } else {
-            internal_err!("Expects default value to have Int64 type")
+    match default_value {
+        Some(v) if v.data_type() == DataType::Int64 => {
+            ScalarValue::try_from_string(v.to_string(), dtype)
         }
-    } else {
-        Ok(ScalarValue::try_from(dtype)?)
+        Some(v) if !v.data_type().is_null() => exec_err!(
+            "Unexpected datatype for default value: {}. Expected: Int64",
+            v.data_type()
+        ),
+        // If None or Null datatype
+        _ => Ok(ScalarValue::try_from(dtype)?),
     }
 }
 

datafusion/sqllogictest/test_files/window.slt

@@ -3990,3 +3990,17 @@ query T
 select arrow_typeof(nth_value(a, 1) over ()) from (select 1 a)
 ----
 Int64
+
+# 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)
+----
+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)
+----
+NULL
+1