diff --git a/expression/integration_test.go b/expression/integration_test.go
index 65284e05a5ec1..06b0a097d538d 100644
--- a/expression/integration_test.go
+++ b/expression/integration_test.go
@@ -9732,6 +9732,35 @@ func (s *testIntegrationSuite) TestGlobalCacheCorrectness(c *C) {
 	tk.MustExec("SET GLOBAL max_connections=151")
 }
 
+func (s *testIntegrationSuite) TestRedundantColumnResolve(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t1, t2")
+	tk.MustExec("create table t1(a int not null)")
+	tk.MustExec("create table t2(a int not null)")
+	tk.MustExec("insert into t1 values(1)")
+	tk.MustExec("insert into t2 values(1)")
+	tk.MustQuery("select a, count(*) from t1 join t2 using (a) group by a").Check(testkit.Rows("1 1"))
+	tk.MustQuery("select a, count(*) from t1 natural join t2 group by a").Check(testkit.Rows("1 1"))
+	err := tk.ExecToErr("select a, count(*) from t1 join t2 on t1.a=t2.a group by a")
+	c.Assert(err.Error(), Equals, "[planner:1052]Column 'a' in field list is ambiguous")
+	tk.MustQuery("select t1.a, t2.a from t1 join t2 using (a) group by t1.a").Check(testkit.Rows("1 1"))
+	err = tk.ExecToErr("select t1.a, t2.a from t1 join t2 using(a) group by a")
+	c.Assert(err.Error(), Equals, "[planner:1052]Column 'a' in group statement is ambiguous")
+	tk.MustQuery("select t2.a from t1 join t2 using (a) group by t1.a").Check(testkit.Rows("1"))
+	tk.MustQuery("select t1.a from t1 join t2 using (a) group by t1.a").Check(testkit.Rows("1"))
+	tk.MustQuery("select t2.a from t1 join t2 using (a) group by t2.a").Check(testkit.Rows("1"))
+	// The test below cannot pass now since we do not infer functional dependencies from filters as MySQL, hence would fail in only_full_group_by check.
+	// tk.MustQuery("select t1.a from t1 join t2 using (a) group by t2.a").Check(testkit.Rows("1"))
+	tk.MustQuery("select count(*) from t1 join t2 using (a) group by t2.a").Check(testkit.Rows("1"))
+	tk.MustQuery("select t2.a from t1 join t2 using (a) group by a").Check(testkit.Rows("1"))
+	tk.MustQuery("select t1.a from t1 join t2 using (a) group by a").Check(testkit.Rows("1"))
+	tk.MustQuery("select * from t1 join t2 using(a)").Check(testkit.Rows("1"))
+	tk.MustQuery("select t1.a, t2.a from t1 join t2 using(a)").Check(testkit.Rows("1 1"))
+	tk.MustQuery("select * from t1 natural join t2").Check(testkit.Rows("1"))
+	tk.MustQuery("select t1.a, t2.a from t1 natural join t2").Check(testkit.Rows("1 1"))
+}
+
 func (s *testIntegrationSuite) TestControlFunctionWithEnumOrSet(c *C) {
 	defer s.cleanEnv(c)
 
diff --git a/expression/simple_rewriter.go b/expression/simple_rewriter.go
index 46486576b9117..ef52bff089401 100644
--- a/expression/simple_rewriter.go
+++ b/expression/simple_rewriter.go
@@ -129,6 +129,12 @@ func FindFieldName(names types.NameSlice, astCol *ast.ColumnName) (int, error) {
 			if idx == -1 {
 				idx = i
 			} else {
+				if names[idx].Redundant || name.Redundant {
+					if !name.Redundant {
+						idx = i
+					}
+					continue
+				}
 				return -1, errNonUniq.GenWithStackByArgs(astCol.String(), "field list")
 			}
 		}
diff --git a/planner/core/logical_plan_builder.go b/planner/core/logical_plan_builder.go
index 51e14463db742..5bb3c34e327be 100644
--- a/planner/core/logical_plan_builder.go
+++ b/planner/core/logical_plan_builder.go
@@ -841,6 +841,7 @@ func (b *PlanBuilder) coalesceCommonColumns(p *LogicalJoin, leftPlan, rightPlan
 	lColumns, rColumns := lsc.Columns, rsc.Columns
 	lNames, rNames := leftPlan.OutputNames().Shallow(), rightPlan.OutputNames().Shallow()
 	if joinTp == ast.RightJoin {
+		leftPlan, rightPlan = rightPlan, leftPlan
 		lNames, rNames = rNames, lNames
 		lColumns, rColumns = rsc.Columns, lsc.Columns
 	}
@@ -938,16 +939,18 @@ func (b *PlanBuilder) coalesceCommonColumns(p *LogicalJoin, leftPlan, rightPlan
 
 	p.SetSchema(expression.NewSchema(schemaCols...))
 	p.names = names
-	if joinTp == ast.RightJoin {
-		leftPlan, rightPlan = rightPlan, leftPlan
-	}
 	// We record the full `rightPlan.Schema` as `redundantSchema` in order to
 	// record the redundant column in `rightPlan` and the output columns order
 	// of the `rightPlan`.
 	// For SQL like `select t1.*, t2.* from t1 left join t2 using(a)`, we can
 	// retrieve the column order of `t2.*` from the `redundantSchema`.
 	p.redundantSchema = expression.MergeSchema(p.redundantSchema, expression.NewSchema(rightPlan.Schema().Clone().Columns...))
-	p.redundantNames = append(p.redundantNames.Shallow(), rightPlan.OutputNames().Shallow()...)
+	p.redundantNames = p.redundantNames.Shallow()
+	for _, name := range rightPlan.OutputNames() {
+		cpyName := *name
+		cpyName.Redundant = true
+		p.redundantNames = append(p.redundantNames, &cpyName)
+	}
 	if joinTp == ast.RightJoin || joinTp == ast.LeftJoin {
 		resetNotNullFlag(p.redundantSchema, 0, p.redundantSchema.Len())
 	}
@@ -2548,6 +2551,7 @@ func (g *gbyResolver) Leave(inNode ast.Node) (ast.Node, bool) {
 			var index int
 			index, g.err = resolveFromSelectFields(v, g.fields, false)
 			if g.err != nil {
+				g.err = ErrAmbiguous.GenWithStackByArgs(v.Name.Name.L, clauseMsg[groupByClause])
 				return inNode, false
 			}
 			if idx >= 0 {
diff --git a/planner/core/logical_plan_test.go b/planner/core/logical_plan_test.go
index ba842b08d5906..553361c47ebcd 100644
--- a/planner/core/logical_plan_test.go
+++ b/planner/core/logical_plan_test.go
@@ -1459,7 +1459,7 @@ func (s *testPlanSuite) TestNameResolver(c *C) {
 		{"select * from t", ""},
 		{"select t.* from t", ""},
 		{"select t2.* from t", "[planner:1051]Unknown table 't2'"},
-		{"select b as a, c as a from t group by a", "[planner:1052]Column 'c' in field list is ambiguous"},
+		{"select b as a, c as a from t group by a", "[planner:1052]Column 'a' in group statement is ambiguous"},
 		{"select 1 as a, b as a, c as a from t group by a", ""},
 		{"select a, b as a from t group by a+1", ""},
 		{"select c, a as c from t order by c+1", ""},
diff --git a/types/field_name.go b/types/field_name.go
index f7d6771321714..ded69170f375e 100644
--- a/types/field_name.go
+++ b/types/field_name.go
@@ -35,6 +35,8 @@ type FieldName struct {
 	// update stmt can write `writeable` column implicitly but cannot use non-public columns explicit.
 	// e.g. update t set a = 10 where b = 10; which `b` is in `writeOnly` state
 	NotExplicitUsable bool
+
+	Redundant bool
 }
 
 const emptyName = "EMPTY_NAME"