diff --git a/datafusion/sql/src/select.rs b/datafusion/sql/src/select.rs
index bbd9d203eb124..fe601bc08d052 100644
--- a/datafusion/sql/src/select.rs
+++ b/datafusion/sql/src/select.rs
@@ -233,9 +233,26 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
             exprs
                 .into_iter()
                 .map(|e| {
+                    // SQL GROUP BY gives an input column precedence over a same-named
+                    // SELECT-list alias. Resolve a bare input identifier before adding the
+                    // projected schema, which would otherwise make the two fields ambiguous.
+                    let input_column_precedence = match &e {
+                        SQLExpr::Identifier(identifier) => base_plan
+                            .schema()
+                            .qualified_field_with_unqualified_name(
+                                &self.ident_normalizer.normalize(identifier.clone()),
+                            )
+                            .is_ok(),
+                        _ => false,
+                    };
+                    let group_by_schema = if input_column_precedence {
+                        base_plan.schema().as_ref()
+                    } else {
+                        &combined_schema
+                    };
                     let group_by_expr = self.sql_expr_to_logical_expr(
                         e,
-                        &combined_schema,
+                        group_by_schema,
                         planner_context,
                     )?;
 
@@ -248,13 +265,18 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
                         resolve_aliases_to_exprs(group_by_expr, &alias_map)?;
                     let group_by_expr =
                         resolve_positions_to_exprs(group_by_expr, &select_exprs)?;
-                    let group_by_expr = normalize_col(group_by_expr, &projected_plan)?;
+                    let normalization_plan = if input_column_precedence {
+                        &base_plan
+                    } else {
+                        &projected_plan
+                    };
+                    let group_by_expr = normalize_col(group_by_expr, normalization_plan)?;
                     self.validate_schema_satisfies_exprs(
                         base_plan.schema(),
                         std::slice::from_ref(&group_by_expr),
                     )?;
                     let (group_by_expr, _) =
-                        group_by_expr.infer_placeholder_types(&combined_schema)?;
+                        group_by_expr.infer_placeholder_types(group_by_schema)?;
                     Ok(group_by_expr)
                 })
                 .collect::<Result<Vec<Expr>>>()?
diff --git a/datafusion/sql/tests/sql_integration.rs b/datafusion/sql/tests/sql_integration.rs
index 7ca7eccdd42b9..85c1eda25d8ee 100644
--- a/datafusion/sql/tests/sql_integration.rs
+++ b/datafusion/sql/tests/sql_integration.rs
@@ -2046,6 +2046,29 @@ fn select_simple_aggregate_with_groupby_can_use_alias() {
     );
 }
 
+#[test]
+fn group_by_input_column_takes_precedence_over_same_named_alias() {
+    let plan = logical_plan("SELECT SUM(id) AS age FROM person GROUP BY age").unwrap();
+    assert_snapshot!(
+        plan,
+        @r"
+    Projection: sum(person.id) AS age
+      Aggregate: groupBy=[[person.age]], aggr=[[sum(person.id)]]
+        TableScan: person
+    "
+    );
+}
+
+#[test]
+fn group_by_alias_conflict_reaches_non_aggregate_validation() {
+    let error = logical_plan("SELECT state AS age FROM person GROUP BY age")
+        .expect_err("state is not grouped");
+    assert_snapshot!(
+        error.strip_backtrace(),
+        @r#"Error during planning: Column in SELECT must be in GROUP BY or an aggregate function: While expanding wildcard, column "person.state" must appear in the GROUP BY clause or must be part of an aggregate function, currently only "person.age" appears in the SELECT clause satisfies this requirement"#
+    );
+}
+
 #[test]
 fn select_simple_aggregate_with_groupby_aggregate_repeated() {
     let sql = "SELECT state, MIN(age), MIN(age) FROM person GROUP BY state";