Fix serde of window lead/lag defaults #20608

datafusion/physical-plan/src/windows/mod.rs

@@ -216,6 +216,18 @@ impl WindowUDFExpr {
     pub fn fun(&self) -> &Arc<WindowUDF> {
         &self.fun
     }
+
+    /// Returns all arguments passed to this window function.
+    ///
+    /// Unlike [`StandardWindowFunctionExpr::expressions`], which returns
+    /// only the expressions that need batch evaluation (and may filter out
+    /// literal offset/default args like those for `lead`/`lag`), this
+    /// method returns the complete, unfiltered argument list. This is
+    /// needed for serialization so that all arguments survive a
+    /// protobuf round-trip.
+    pub fn args(&self) -> &[Arc<dyn PhysicalExpr>] {
+        &self.args
+    }
 }
 
 impl StandardWindowFunctionExpr for WindowUDFExpr {

datafusion/proto/src/physical_plan/to_proto.rs

@@ -102,7 +102,7 @@ pub fn serialize_physical_window_expr(
     codec: &dyn PhysicalExtensionCodec,
 ) -> Result<protobuf::PhysicalWindowExprNode> {
     let expr = window_expr.as_any();
-    let args = window_expr.expressions().to_vec();
+    let mut args = window_expr.expressions().to_vec();
     let window_frame = window_expr.get_window_frame();
 
     let (window_function, fun_definition) = if let Some(plain_aggr_window_expr) =
@@ -129,6 +129,7 @@ pub fn serialize_physical_window_expr(
         {
             let mut buf = Vec::new();
             codec.try_encode_udwf(expr.fun(), &mut buf)?;
+            args = expr.args().to_vec();
             (
                 physical_window_expr_node::WindowFunction::UserDefinedWindowFunction(
                     expr.fun().name().to_string(),

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -1710,3 +1710,48 @@ fn roundtrip_call_null_scalar_struct_dict() {
     );
     roundtrip_test(filter).expect("roundtrip");
 }
+
+/// Tests that `lead` window function with offset and default value args
+/// survives a protobuf round-trip. This is a regression test for a bug
+/// where `expressions()` (used during serialization) returns only the
+/// column expression for lead/lag, silently dropping the offset and
+/// default value literal args.
+#[test]
+fn roundtrip_lead_with_default_value() -> Result<()> {
+    use datafusion::functions_window::lead_lag::lead_udwf;
+
+    let field_a = Field::new("a", DataType::Int64, false);
+    let field_b = Field::new("b", DataType::Int64, false);
+    let schema = Arc::new(Schema::new(vec![field_a, field_b]));
+
+    // lead(a, 2, 42) — column a, offset 2, default value 42
+    let lead_window = create_udwf_window_expr(
+        &lead_udwf(),
+        &[col("a", &schema)?, lit(2i64), lit(42i64)],
+        schema.as_ref(),
+        "test lead with default".to_string(),
+        false,
+    )?;
+
+    let udwf_expr = Arc::new(StandardWindowExpr::new(
+        lead_window,
+        &[col("b", &schema)?],
+        &LexOrdering::new(vec![PhysicalSortExpr {
+            expr: col("a", &schema)?,
+            options: SortOptions {
+                descending: false,
+                nulls_first: false,
+            },
+        }]),
+        Arc::new(WindowFrame::new(None)),
+    ));
+
+    let input = Arc::new(EmptyExec::new(schema.clone()));
+
+    roundtrip_test(Arc::new(BoundedWindowAggExec::try_new(
+        vec![udwf_expr],
+        input,
+        InputOrderMode::Sorted,
+        true,
+    )?))
+}