@@ -289,10 +289,10 @@ mod tests {
 
         insta::assert_snapshot!(batches_to_string(&results),@r"
         +---------+---------+---------+
-        | column2 | column1 | column3 |
+        | column1 | column2 | column3 |
         +---------+---------+---------+
-        | 2       | 1       | 3       |
-        | 5       | 4       | 6       |
+        | 1       | 2       | 3       |
+        | 4       | 5       | 6       |
         +---------+---------+---------+
         ");
 
 
@@ -378,7 +378,7 @@ async fn optimized_duckdb_unparse_qualifies_nested_passthrough_column() -> Resul
     // `o` (which is only the base-table alias one level deeper). The bug emitted
     // `"o"."order_id"` inside that derived table; the fix emits a bare column.
     let expected = concat!(
-        r#"SELECT "o"."order_id", "o"."discount_pct_2" "#,
+        r#"SELECT "oi"."order_id", "o"."discount_pct_2" "#,
         r#"FROM "warehouse"."main"."order_items" AS "oi" "#,
         r#"INNER JOIN (SELECT "order_id", "#,
         r#"CASE WHEN "__common_expr_1" IS NOT NULL "#,
 
@@ -22,14 +22,15 @@ use std::collections::HashSet;
 use std::fmt::Debug;
 use std::sync::Arc;
 
+use crate::conditional_expressions::CaseBuilder;
 use crate::expr::{Alias, Sort, Unnest};
 use crate::logical_plan::Projection;
-use crate::{Expr, ExprSchemable, LogicalPlan};
+use crate::{Expr, ExprSchemable, LogicalPlan, LogicalPlanBuilder};
 
 use datafusion_common::TableReference;
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
-use datafusion_common::{Column, DFSchema, Result};
+use datafusion_common::{Column, DFSchema, JoinType, Result};
 
 mod guarantees;
 pub use guarantees::GuaranteeRewriter;
@@ -65,6 +66,10 @@ pub trait FunctionRewrite: Debug {
 }
 
 /// Recursively normalize all [`Column`] expressions in the `expr` expression tree.
+///
+/// An unqualified reference to the merged key of a `USING` / `NATURAL` join is
+/// resolved to a reserved internal merged-key expression via
+/// [`merged_using_key_or_column`].
 pub fn normalize_col(expr: Expr, plan: &LogicalPlan) -> Result<Expr> {
     ColumnNormalizer::new(plan).normalize(expr)
 }
@@ -74,13 +79,15 @@ pub fn normalize_col(expr: Expr, plan: &LogicalPlan) -> Result<Expr> {
 pub(crate) struct ColumnNormalizer<'a> {
     plan: &'a LogicalPlan,
     context: Option<(Vec<&'a DFSchema>, Vec<HashSet<Column>>)>,
+    merged_keys: Option<Vec<(Column, Column, JoinType)>>,
 }
 
 impl<'a> ColumnNormalizer<'a> {
     pub(crate) fn new(plan: &'a LogicalPlan) -> Self {
         Self {
             plan,
             context: None,
+            merged_keys: None,
         }
     }
 
@@ -105,15 +112,76 @@ impl<'a> ColumnNormalizer<'a> {
 
     pub(crate) fn normalize(&mut self, expr: Expr) -> Result<Expr> {
         expr.transform(|expr| match expr {
-            Expr::Column(column) => self
-                .normalize_column(column)
-                .map(|column| Transformed::yes(Expr::Column(column))),
+            Expr::Column(column) => {
+                let was_unqualified = column.relation.is_none();
+                let column = self.normalize_column(column)?;
+                let expr = if was_unqualified {
+                    let merged_keys = match &mut self.merged_keys {
+                        Some(keys) => keys,
+                        keys @ None => keys.insert(self.plan.using_key_pairs()?),
+                    };
+                    merged_using_key_or_column(column, true, merged_keys)?
+                } else {
+                    Expr::Column(column)
+                };
+                Ok(Transformed::yes(expr))
+            }
             _ => Ok(Transformed::no(expr)),
         })
         .data()
     }
 }
 
+fn alias_merged_key(expr: Expr, name: &str) -> Expr {
+    expr.alias_qualified(
+        Some(TableReference::bare(
+            LogicalPlanBuilder::MERGED_KEY_QUALIFIER,
+        )),
+        name,
+    )
+}
+
+/// Resolve a normalized column to itself, or -- when it is the merged key of a
+/// `USING` / `NATURAL` join referenced *unqualified* -- to the merged key
+/// expression with a reserved internal qualifier.
+///
+/// `merged_keys` carries the `(left, right, join_type)` triples of the plan's
+/// USING / NATURAL joins. The merged key resolves to the left key for
+/// left-preserving joins, the right key for right-preserving joins, and
+/// `COALESCE(left, right)` for `FULL`. The COALESCE is built as
+/// `CASE WHEN left IS NOT NULL THEN left ELSE right END` (the form `coalesce`
+/// is simplified to, buildable here without depending on the functions crate).
+pub fn merged_using_key_or_column(
+    col: Column,
+    was_unqualified: bool,
+    merged_keys: &[(Column, Column, JoinType)],
+) -> Result<Expr> {
+    if was_unqualified
+        && let Some((l, r, join_type)) =
+            merged_keys.iter().find(|(l, r, _)| l == &col || r == &col)
+    {
+        let name = col.name.clone();
+        let expr = match join_type {
+            // The left key is NULL-padded on right-only rows; the right key is
+            // always present, so the merged key is just the right column.
+            JoinType::Right | JoinType::RightSemi | JoinType::RightAnti => {
+                Expr::Column(r.clone())
+            }
+            // Either side may be NULL-padded; coalesce to the present one.
+            JoinType::Full => CaseBuilder::new(
+                None,
+                vec![Expr::Column(l.clone()).is_not_null()],
+                vec![Expr::Column(l.clone())],
+                Some(Box::new(Expr::Column(r.clone()))),
+            )
+            .end()?,
+            _ => Expr::Column(l.clone()),
+        };
+        return Ok(alias_merged_key(expr, &name));
+    }
+    Ok(Expr::Column(col))
+}
+
 /// See [`Column::normalize_with_schemas_and_ambiguity_check`] for usage
 pub fn normalize_col_with_schemas_and_ambiguity_check(
     expr: Expr,
@@ -136,9 +204,9 @@ pub fn normalize_col_with_schemas_and_ambiguity_check(
     expr.transform(|expr| {
         Ok({
             if let Expr::Column(c) = expr {
-                let col =
-                    c.normalize_with_schemas_and_ambiguity_check(schemas, using_columns)?;
-                Transformed::yes(Expr::Column(col))
+                Transformed::yes(Expr::Column(
+                    c.normalize_with_schemas_and_ambiguity_check(schemas, using_columns)?,
+                ))
             } else {
                 Transformed::no(expr)
             }
@@ -561,18 +629,24 @@ mod test {
         let right =
             crate::logical_plan::table_scan(Some("right"), &schema, None)?.build()?;
         let plan = crate::logical_plan::table_scan(Some("left"), &schema, None)?
-            .join_using(right, crate::JoinType::Inner, vec![Column::from_name("a")])?
+            .join_using(right, JoinType::Inner, vec![Column::from_name("a")])?
             .build()?;
+        let merged_key = col("left.a").alias_qualified(
+            Some(TableReference::bare(
+                LogicalPlanBuilder::MERGED_KEY_QUALIFIER,
+            )),
+            "a",
+        );
         assert_eq!(
             super::normalize_cols([col("a"), col("a") + col("right.a")], &plan)?,
-            vec![col("left.a"), col("left.a") + col("right.a")]
+            vec![merged_key.clone(), merged_key.clone() + col("right.a")]
         );
         let projected = LogicalPlanBuilder::from(plan.clone())
             .project([col("a").alias("key"), col("left.b").alias("value")])?
             .build()?;
         assert_eq!(
             projected.expressions(),
-            vec![col("left.a").alias("key"), col("left.b").alias("value")]
+            vec![merged_key.alias("key"), col("left.b").alias("value")]
         );
         let err = super::normalize_cols([col("a"), col("b"), col("missing")], &plan)
             .unwrap_err();
@@ -589,7 +663,7 @@ mod test {
         let right =
             crate::logical_plan::table_scan(Some("right"), &schema, None)?.build()?;
         let mut plan = crate::logical_plan::table_scan(Some("left"), &schema, None)?
-            .join_using(right, crate::JoinType::Inner, vec![Column::from_name("a")])?
+            .join_using(right, JoinType::Inner, vec![Column::from_name("a")])?
             .build()?;
         // Invalid USING keys must only be inspected when an unqualified column
         // needs normalization, just as with a single expression.
 
@@ -132,6 +132,10 @@ pub struct LogicalPlanBuilder {
 }
 
 impl LogicalPlanBuilder {
+    /// Reserved internal qualifier for USING / NATURAL join merged keys.
+    /// A real relation carrying this qualifier would clash.
+    pub const MERGED_KEY_QUALIFIER: &str = "__datafusion_merged_key_qualifier__";
+
     /// Create a builder from an existing plan
     pub fn new(plan: LogicalPlan) -> Self {
         Self {
@@ -854,12 +858,10 @@ impl LogicalPlanBuilder {
         // remove pushed down sort columns
         let new_expr = schema.columns().into_iter().map(Expr::Column).collect();
 
+        let input = Arc::unwrap_or_clone(self.plan);
+
         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)?,
 
@@ -546,6 +546,35 @@ impl LogicalPlan {
         Ok(using_columns)
     }
 
+    /// Returns the `(left, right, join_type)` join-key column triples of every
+    /// `USING` / `NATURAL` join in this plan.
+    pub fn using_key_pairs(
+        &self,
+    ) -> Result<Vec<(Column, Column, JoinType)>, DataFusionError> {
+        let mut triples = vec![];
+
+        self.apply_with_subqueries(|plan| {
+            if let LogicalPlan::Join(Join {
+                join_constraint: JoinConstraint::Using,
+                join_type,
+                on,
+                ..
+            }) = plan
+            {
+                for (l, r) in on {
+                    if let (Some(l), Some(r)) =
+                        (l.get_as_join_column(), r.get_as_join_column())
+                    {
+                        triples.push((l.to_owned(), r.to_owned(), *join_type));
+                    }
+                }
+            }
+            Ok(TreeNodeRecursion::Continue)
+        })?;
+
+        Ok(triples)
+    }
+
     /// returns the first output expression of this `LogicalPlan` node.
     pub fn head_output_expr(&self) -> Result<Option<Expr>> {
         match self {
 
@@ -21,8 +21,10 @@ use std::cmp::Ordering;
 use std::collections::{BTreeSet, HashSet};
 use std::sync::Arc;
 
+use crate::conditional_expressions::CaseBuilder;
 use crate::expr::{Alias, Sort, WildcardOptions, WindowFunctionParams};
 use crate::expr_rewriter::strip_outer_reference;
+use crate::logical_plan::JoinType;
 use crate::{
     BinaryExpr, Expr, ExprSchemable, Filter, GroupingSet, LogicalPlan, Operator, and,
 };
@@ -409,37 +411,54 @@ fn get_exprs_except_skipped(
 /// joins). An unqualified wildcard should include each USING column only once.
 /// This function returns the duplicate columns that should be excluded.
 fn exclude_using_columns(plan: &LogicalPlan) -> Result<HashSet<Column>> {
-    let output_columns: HashSet<_> = plan.schema().columns().iter().cloned().collect();
+    let output_columns = plan.schema().columns();
     let mut excluded = HashSet::new();
     for cols in plan.using_columns()? {
         // `using_columns()` returns join columns from both sides regardless of
-        // the join type. For semi/anti joins, only one side's columns appear in
-        // the output schema. Filter to output columns so that columns from the
-        // non-output side don't participate in the deduplication process below
-        // and displace real output columns.
-        let mut cols: Vec<_> = cols
-            .into_iter()
-            .filter(|c| output_columns.contains(c))
-            .collect();
-
-        // Sort so we keep the same qualified column, regardless of HashSet
-        // iteration order.
-        cols.sort();
-
-        // Keep only one column per name from the columns set, adding any
-        // duplicates to the excluded set.
+        // the join type. Visit only output columns, in schema order, so the
+        // surviving key keeps its position in an unqualified wildcard.
         let mut seen_names = HashSet::new();
-        for col in cols {
-            if seen_names.contains(col.name.as_str()) {
-                excluded.insert(col); // exclude columns with already seen name
-            } else {
-                seen_names.insert(col.name.clone()); // mark column name as seen
+        for col in output_columns.iter().filter(|col| cols.contains(*col)) {
+            if !seen_names.insert(&col.name) {
+                excluded.insert(col.clone());
             }
         }
     }
     Ok(excluded)
 }
 
+fn merged_using_key_for_wildcard(
+    col: Column,
+    merged_keys: &[(Column, Column, JoinType)],
+) -> Result<Expr> {
+    let Some((l, r, join_type)) =
+        merged_keys.iter().find(|(l, r, _)| l == &col || r == &col)
+    else {
+        return Ok(Expr::Column(col));
+    };
+
+    let (expr, visible_col) = match join_type {
+        JoinType::Right | JoinType::RightSemi | JoinType::RightAnti => {
+            (Expr::Column(r.clone()), r.clone())
+        }
+        JoinType::Full => CaseBuilder::new(
+            None,
+            vec![Expr::Column(l.clone()).is_not_null()],
+            vec![Expr::Column(l.clone())],
+            Some(Box::new(Expr::Column(r.clone()))),
+        )
+        .end()
+        .map(|expr| (expr, col.clone()))?,
+        _ => (Expr::Column(l.clone()), l.clone()),
+    };
+
+    if expr == Expr::Column(visible_col.clone()) {
+        Ok(expr)
+    } else {
+        Ok(expr.alias_qualified(visible_col.relation, visible_col.name))
+    }
+}
+
 /// Resolves an `Expr::Wildcard` to a collection of `Expr::Column`'s.
 pub fn expand_wildcard(
     schema: &DFSchema,
@@ -459,7 +478,21 @@ pub fn expand_wildcard(
     };
     // Add each excluded `Column` to columns_to_skip
     columns_to_skip.extend(excluded_columns);
-    Ok(get_exprs_except_skipped(schema, &columns_to_skip))
+    let exprs = get_exprs_except_skipped(schema, &columns_to_skip);
+
+    // Resolve the surviving USING / NATURAL key column to the merged key value,
+    // while preserving the wildcard-visible field qualifier.
+    let merged_keys = plan.using_key_pairs()?;
+    if merged_keys.is_empty() {
+        return Ok(exprs);
+    }
+    exprs
+        .into_iter()
+        .map(|expr| match expr {
+            Expr::Column(col) => merged_using_key_for_wildcard(col, &merged_keys),
+            other => Ok(other),
+        })
+        .collect()
 }
 
 /// Resolves an `Expr::Wildcard` to a collection of qualified `Expr::Column`'s.
 
@@ -2035,7 +2035,7 @@ mod tests {
         assert_snapshot!(
             optimized_plan.clone(),
             @r"
-        Projection: test.a, test.b
+        Projection: test.a AS a, test.b
           Left Join: Using test.a = test2.a
             TableScan: test projection=[a, b]
             TableScan: test2 projection=[a]