Fix blocking issue and clean up projection logic

Remove the EXPLAIN-time UPDATE ... FROM rejection in
statement.rs to allow the SQL planner to expose the joined
logical plan. Adjust regression test in sql_integration.rs
to assert the Explain -> Dml(Update) plan shape.

Consolidate duplicated projection-walking logic in
physical_planner.rs by using a shared helper function
for extract_update_assignments(). This simplifies
identity-check and qualifier-normalization rules.

Author: kosiew

datafusion/core/src/physical_planner.rs

@@ -2277,42 +2277,22 @@ fn extract_update_assignments(
 
     // Find the top-level projection
     if let LogicalPlan::Projection(projection) = input.as_ref() {
-        for expr in &projection.expr {
-            if let Expr::Alias(alias) = expr {
-                // The alias name is the column name being updated
-                // The inner expression is the new value
-                let column_name = alias.name.clone();
-                // Only include if it's not just a column reference to itself
-                // (those are columns that aren't being updated)
-                if !is_identity_assignment(&alias.expr, &column_name, &target_refs) {
-                    let assignment_expr = normalize_update_assignment_expr(
-                        (*alias.expr).clone(),
-                        strip_qualifiers,
-                    )?;
-                    assignments.push((column_name, assignment_expr));
-                }
-            }
-        }
+        append_update_assignments(
+            &mut assignments,
+            projection,
+            &target_refs,
+            strip_qualifiers,
+        )?;
     } else {
         // Try to find projection deeper in the plan
         input.apply(|node| {
             if let LogicalPlan::Projection(projection) = node {
-                for expr in &projection.expr {
-                    if let Expr::Alias(alias) = expr {
-                        let column_name = alias.name.clone();
-                        if !is_identity_assignment(
-                            &alias.expr,
-                            &column_name,
-                            &target_refs,
-                        ) {
-                            let assignment_expr = normalize_update_assignment_expr(
-                                (*alias.expr).clone(),
-                                strip_qualifiers,
-                            )?;
-                            assignments.push((column_name, assignment_expr));
-                        }
-                    }
-                }
+                append_update_assignments(
+                    &mut assignments,
+                    projection,
+                    &target_refs,
+                    strip_qualifiers,
+                )?;
                 return Ok(TreeNodeRecursion::Stop);
             }
             Ok(TreeNodeRecursion::Continue)
@@ -2322,6 +2302,31 @@ fn extract_update_assignments(
     Ok(assignments)
 }
 
+fn append_update_assignments(
+    assignments: &mut Vec<(String, Expr)>,
+    projection: &Projection,
+    target_refs: &[TableReference],
+    strip_qualifiers: bool,
+) -> Result<()> {
+    for expr in &projection.expr {
+        if let Expr::Alias(alias) = expr {
+            // The alias name is the column name being updated
+            // The inner expression is the new value
+            let column_name = alias.name.clone();
+            // Only include if it's not just a column reference to itself
+            // (those are columns that aren't being updated)
+            if !is_identity_assignment(&alias.expr, &column_name, target_refs) {
+                let assignment_expr = normalize_update_assignment_expr(
+                    (*alias.expr).clone(),
+                    strip_qualifiers,
+                )?;
+                assignments.push((column_name, assignment_expr));
+            }
+        }
+    }
+    Ok(())
+}
+
 fn collect_target_refs(
     input: &Arc<LogicalPlan>,
     target: &TableReference,

datafusion/sql/src/statement.rs

@@ -32,7 +32,6 @@ use crate::utils::normalize_ident;
 use arrow::datatypes::{Field, FieldRef, Fields};
 use datafusion_common::error::_plan_err;
 use datafusion_common::parsers::CompressionTypeVariant;
-use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion};
 use datafusion_common::{
     Column, Constraint, Constraints, DFSchema, DFSchemaRef, DataFusionError, Result,
     ScalarValue, SchemaError, SchemaReference, TableReference, ToDFSchema, exec_err,
@@ -1913,12 +1912,6 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
         statement: DFStatement,
     ) -> Result<LogicalPlan> {
         let plan = self.statement_to_plan(statement)?;
-        // TODO: remove this guard once UPDATE ... FROM routing via
-        // TableProvider::update_from(...) and joined assignment handling land.
-        // See https://github.com/apache/datafusion/issues/19950.
-        if update_uses_joined_input(&plan)? {
-            return not_impl_err!("UPDATE ... FROM is not supported");
-        }
         if matches!(plan, LogicalPlan::Explain(_)) {
             return plan_err!("Nested EXPLAINs are not supported");
         }
@@ -2573,24 +2566,3 @@ ON p.function_name = r.routine_name
         }
     }
 }
-
-fn update_uses_joined_input(plan: &LogicalPlan) -> Result<bool> {
-    let LogicalPlan::Dml(DmlStatement {
-        op: WriteOp::Update,
-        input,
-        ..
-    }) = plan
-    else {
-        return Ok(false);
-    };
-
-    let mut has_join = false;
-    input.apply(|node| {
-        if matches!(node, LogicalPlan::Join(_)) {
-            has_join = true;
-            return Ok(TreeNodeRecursion::Stop);
-        }
-        Ok(TreeNodeRecursion::Continue)
-    })?;
-    Ok(has_join)
-}

datafusion/sql/tests/sql_integration.rs

@@ -761,13 +761,21 @@ fn plan_update_from_with_aliases() {
 }
 
 #[test]
-fn explain_update_from_is_rejected() {
+fn plan_explain_update_from() {
     let sql = "EXPLAIN UPDATE t1 SET b = t2.b, c = t2.a, d = 1 \
                FROM t2 WHERE t1.a = t2.a AND t1.b > 'foo' AND t2.c > 1.0";
-    let err = logical_plan(sql).expect_err("EXPLAIN UPDATE ... FROM should fail");
+    let plan = logical_plan(sql).unwrap();
     assert_snapshot!(
-        err.strip_backtrace(),
-        @r#"This feature is not implemented: UPDATE ... FROM is not supported"#
+        plan,
+        @r#"
+    Explain
+      Dml: op=[Update] table=[t1]
+        Projection: t1.a AS a, t2.b AS b, CAST(t2.a AS Float64) AS c, CAST(Int64(1) AS Int32) AS d
+          Filter: t1.a = t2.a AND t1.b > Utf8("foo") AND t2.c > Float64(1)
+            Cross Join:
+              TableScan: t1
+              TableScan: t2
+    "#
     );
 }