@@ -53,6 +53,7 @@ use arrow::datatypes::{DataType, Field, FieldRef, Fields, Schema, SchemaRef};
 use datafusion_common::display::ToStringifiedPlan;
 use datafusion_common::file_options::file_type::FileType;
 use datafusion_common::metadata::FieldMetadata;
+use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
 use datafusion_common::{
     Column, Constraints, DFSchema, DFSchemaRef, NullEquality, Result, ScalarValue,
     TableReference, ToDFSchema, UnnestOptions, exec_err,
@@ -130,6 +131,11 @@ pub struct LogicalPlanBuilder {
 }
 
 impl LogicalPlanBuilder {
+    /// Reserved name prefix for a FULL USING / NATURAL join merged key that has
+    /// been renamed to dodge an unqualified/qualified schema collision during
+    /// sort push-down. A real column carrying this prefix would clash.
+    const MERGED_KEY_NAME_PREFIX: &str = "__datafusion_merged_key_";
+
     /// Create a builder from an existing plan
     pub fn new(plan: LogicalPlan) -> Self {
         Self {
@@ -819,6 +825,97 @@ impl LogicalPlanBuilder {
         self.sort_with_limit(sorts, None)
     }
 
+    /// Internal name for a FULL USING / NATURAL join merged key that has been
+    /// renamed to dodge an unqualified/qualified schema collision during sort
+    /// push-down. See [`Self::sort_with_limit`].
+    fn synthetic_merged_name(name: &str) -> String {
+        format!("{}{name}", Self::MERGED_KEY_NAME_PREFIX)
+    }
+
+    /// Unqualified projection output fields that collide with a *qualified*
+    /// missing sort column of the same name. In practice these are the
+    /// `COALESCE` merged keys of FULL USING / NATURAL joins, exposed
+    /// unqualified, which `ORDER BY <qualified key>` would otherwise drag into
+    /// an illegal `{ k, a.k }` projection schema. They are renamed to a
+    /// synthetic name before folding; any unqualified reference to them in the
+    /// sort itself is rewritten to match (see [`Self::rewrite_sort_merged_keys`]).
+    fn renamable_merged_keys(
+        schema: &DFSchemaRef,
+        missing_cols: &IndexSet<Column>,
+    ) -> HashSet<String> {
+        let qualified_missing: HashSet<&str> = missing_cols
+            .iter()
+            .filter(|c| c.relation.is_some())
+            .map(|c| c.name.as_str())
+            .collect();
+        if qualified_missing.is_empty() {
+            return HashSet::new();
+        }
+
+        schema
+            .columns()
+            .into_iter()
+            .filter(|c| {
+                c.relation.is_none() && qualified_missing.contains(c.name.as_str())
+            })
+            .map(|c| c.name)
+            .collect()
+    }
+
+    /// Re-alias a renamable merged key output field to its
+    /// [`Self::synthetic_merged_name`] so it no longer collides with the
+    /// qualified sort column folded alongside it.
+    fn rename_merged_key(expr: Expr, renamable: &HashSet<String>) -> Expr {
+        match expr {
+            Expr::Alias(mut alias)
+                if alias.relation.is_none() && renamable.contains(&alias.name) =>
+            {
+                alias.name = Self::synthetic_merged_name(&alias.name);
+                Expr::Alias(alias)
+            }
+            Expr::Column(column)
+                if column.relation.is_none() && renamable.contains(&column.name) =>
+            {
+                let name = Self::synthetic_merged_name(&column.name);
+                Expr::Column(column).alias(name)
+            }
+            other => other,
+        }
+    }
+
+    /// Rewrite unqualified references to a renamed merged key inside the sort
+    /// expressions to its [`Self::synthetic_merged_name`], so a sort that also
+    /// orders by the merged key (e.g. `ORDER BY a.k, k`) keeps resolving after
+    /// the projection field has been renamed.
+    fn rewrite_sort_merged_keys(
+        sorts: Vec<SortExpr>,
+        renamable: &HashSet<String>,
+    ) -> Result<Vec<SortExpr>> {
+        sorts
+            .into_iter()
+            .map(|sort| {
+                let expr = sort
+                    .expr
+                    .clone()
+                    .transform(|e| {
+                        Ok(match e {
+                            Expr::Column(c)
+                                if c.relation.is_none()
+                                    && renamable.contains(&c.name) =>
+                            {
+                                Transformed::yes(Expr::Column(Column::new_unqualified(
+                                    Self::synthetic_merged_name(&c.name),
+                                )))
+                            }
+                            other => Transformed::no(other),
+                        })
+                    })
+                    .data()?;
+                Ok(sort.with_expr(expr))
+            })
+            .collect()
+    }
+
     /// Apply a sort
     pub fn sort_with_limit(
         self,
@@ -852,15 +949,63 @@ impl LogicalPlanBuilder {
             })));
         }
 
+        // A qualified sort column folded into the projection can collide with an
+        // unqualified output field of the same name: the COALESCE merged key of a
+        // FULL USING / NATURAL join is exposed unqualified, so `ORDER BY a.k`
+        // dragging the qualified `a.k` into the `SELECT k` projection would build
+        // an illegal `{ k, a.k }` schema. Rename such merged keys to a unique
+        // internal name before folding, and restore the original name below in
+        // the wrapper projection.
+        let renamable = match self.plan.as_ref() {
+            LogicalPlan::Projection(_) => {
+                Self::renamable_merged_keys(schema, &missing_cols)
+            }
+            _ => HashSet::new(),
+        };
+
+        // Keep the sort's own references to a renamed merged key resolving by
+        // rewriting them to the same synthetic name (e.g. `ORDER BY a.k, k`,
+        // where `k` is the merged key being renamed to dodge the `a.k` clash).
+        let sorts = if renamable.is_empty() {
+            sorts
+        } else {
+            Self::rewrite_sort_merged_keys(sorts, &renamable)?
+        };
+
         // remove pushed down sort columns
-        let new_expr = schema.columns().into_iter().map(Expr::Column).collect();
+        let new_expr = schema
+            .columns()
+            .into_iter()
+            .map(|c| {
+                if c.relation.is_none() && renamable.contains(&c.name) {
+                    Expr::Column(Column::new_unqualified(Self::synthetic_merged_name(
+                        &c.name,
+                    )))
+                    .alias(c.name)
+                } else {
+                    Expr::Column(c)
+                }
+            })
+            .collect();
+
+        let input = if renamable.is_empty() {
+            Arc::unwrap_or_clone(self.plan)
+        } else {
+            match Arc::unwrap_or_clone(self.plan) {
+                LogicalPlan::Projection(Projection { expr, input, .. }) => {
+                    let expr = expr
+                        .into_iter()
+                        .map(|e| Self::rename_merged_key(e, &renamable))
+                        .collect::<Vec<_>>();
+                    LogicalPlan::Projection(Projection::try_new(expr, input)?)
+                }
+                // `renamable` is only non-empty for a Projection (checked above).
+                other => other,
+            }
+        };
 
         let is_distinct = false;
-        let plan = Self::add_missing_columns(
-            Arc::unwrap_or_clone(self.plan),
-            &missing_cols,
-            is_distinct,
-        )?;
+        let plan = Self::add_missing_columns(input, &missing_cols, is_distinct)?;
 
         let sort_plan = LogicalPlan::Sort(Sort {
             expr: normalize_sorts(sorts, &plan)?,
 
@@ -220,7 +220,20 @@ pub(super) fn rewrite_plan_for_sort_on_non_projected_fields(
         })
         .collect::<Vec<_>>();
 
-    let mut collects = p.expr.clone();
+    let mut collects = p
+        .expr
+        .iter()
+        .map(|e| match e {
+            // A pure rename `Column(x) AS out` compares as its underlying
+            // column `x`, so an outer Projection that only renames an inner
+            // column (e.g. the restored merged key of a FULL USING / NATURAL
+            // join) still matches and can be flattened instead of nested.
+            Expr::Alias(alias) if matches!(alias.expr.as_ref(), Expr::Column(_)) => {
+                alias.expr.as_ref().clone()
+            }
+            _ => e.clone(),
+        })
+        .collect::<Vec<_>>();
     for sort in &sort.expr {
         // Strip aliases from sort expressions so the comparison matches
         // the inner Projection's raw expressions. The optimizer may add
@@ -250,7 +263,19 @@ pub(super) fn rewrite_plan_for_sort_on_non_projected_fields(
         let new_exprs = p
             .expr
             .iter()
-            .map(|e| map.get(e).unwrap_or(e).clone())
+            .map(|e| {
+                // For a pure rename `Column(x) AS out`, inline the inner
+                // definition of `x` and re-apply the outer output name, so the
+                // flattened SELECT keeps `out` as the column name.
+                if let Expr::Alias(alias) = e
+                    && matches!(alias.expr.as_ref(), Expr::Column(_))
+                    && let Some(inner) = map.get(alias.expr.as_ref())
+                {
+                    inner.clone().unalias().alias(alias.name.clone())
+                } else {
+                    map.get(e).unwrap_or(e).clone()
+                }
+            })
             .collect::<Vec<_>>();
 
         // The inner Projection may define aliases that the Sort references
 
@@ -46,6 +46,7 @@ use datafusion_sql::unparser::dialect::{
 };
 use datafusion_sql::unparser::{Unparser, expr_to_sql, plan_to_sql};
 use insta::assert_snapshot;
+use rstest::rstest;
 use sqlparser::ast::Statement;
 use std::hash::Hash;
 use std::ops::Add;
@@ -260,6 +261,48 @@ fn roundtrip_statement() -> Result<()> {
     Ok(())
 }
 
+/// Regression for https://github.com/apache/datafusion/issues/22881.
+///
+/// A FULL JOIN USING / NATURAL merged key is `COALESCE(left, right)`, exposed
+/// as an *unqualified* column. Selecting that merged key while ordering by a
+/// *qualified* key of the same join (`ORDER BY ta.id`) must both plan and
+/// survive an unparser round-trip (`plan -> SQL -> plan`). Earlier fixes either
+/// errored at plan time or produced SQL that could not be re-planned. `USING`
+/// and `NATURAL` are covered separately since the unparser renders them
+/// differently.
+#[rstest]
+#[case::full_join_using(
+    "SELECT id FROM (SELECT j1_id AS id FROM j1) ta \
+     FULL JOIN (SELECT j2_id AS id FROM j2) tb USING (id) \
+     ORDER BY ta.id"
+)]
+#[case::natural_full_join(
+    "SELECT id FROM (SELECT j1_id AS id FROM j1) ta \
+     NATURAL FULL JOIN (SELECT j2_id AS id FROM j2) tb \
+     ORDER BY ta.id"
+)]
+fn roundtrip_full_join_merged_key_order_by_qualified(#[case] query: &str) -> Result<()> {
+    let dialect = GenericDialect {};
+    let statement = Parser::new(&dialect)
+        .try_with_sql(query)?
+        .parse_statement()?;
+
+    let context = MockContextProvider {
+        state: MockSessionState::default(),
+    };
+    let sql_to_rel = SqlToRel::new(&context);
+    let plan = sql_to_rel.sql_statement_to_plan(statement).unwrap();
+
+    let roundtrip_statement = plan_to_sql(&plan)?;
+    let plan_roundtrip = sql_to_rel
+        .sql_statement_to_plan(roundtrip_statement)
+        .unwrap();
+
+    assert_eq!(plan, plan_roundtrip);
+
+    Ok(())
+}
+
 #[test]
 fn roundtrip_crossjoin() -> Result<()> {
     let query = "select j1.j1_id, j2.j2_string from j1, j2";
 
@@ -119,6 +119,48 @@ SELECT coalesce(a.k, b.k) AS k FROM a FULL JOIN b ON a.k = b.k ORDER BY k
 3
 4
 
+##########
+# Regression: ORDER BY a *qualified* key while selecting the FULL merged key.
+# The merged k is COALESCE(a.k, b.k), exposed unqualified; ORDER BY a.k folds
+# the qualified a.k into the same projection. This must plan (it regressed to a
+# schema-ambiguity error) and sort by a.k, not by the merged key: the right-only
+# row (a.k IS NULL) sorts last, the rest descend by a.k -- so the output differs
+# from ORDER BY k DESC (which would be 4, 3, 2, 1).
+##########
+
+query I nosort
+SELECT k FROM a FULL JOIN b USING (k) ORDER BY a.k DESC NULLS LAST
+----
+3
+2
+1
+4
+
+query I nosort
+SELECT k FROM a NATURAL FULL JOIN b ORDER BY a.k DESC NULLS LAST
+----
+3
+2
+1
+4
+
+##########
+# Regression: a *qualified* key and the *unqualified* merged key together in one
+# ORDER BY (`ORDER BY a.k DESC NULLS LAST, k`). On main this was legal (the merged
+# key resolved to a.k, so both sort keys were a.k); the merged-key fix must keep
+# it legal -- renaming the merged key to dodge the `{k, a.k}` collision must also
+# rewrite the unqualified `k` reference in the sort. a.k is unique here so the
+# secondary `k` never breaks a tie, but the shape still exercises that path.
+##########
+
+query I nosort
+SELECT k FROM a FULL JOIN b USING (k) ORDER BY a.k DESC NULLS LAST, k
+----
+3
+2
+1
+4
+
 ##########
 # Wildcard over the join: the merged key must appear once and carry the right
 # value. (commit 2 target -- the materialized merged column must dedupe under *)