planner/core: fix redundant USING-column binding in ALL subquery

docs/note/planner/README.md

@@ -2,3 +2,4 @@
 
 - `rule`: rule-related notes and decisions.
 - `plan_cache_notes.md`: plan cache and binding matching notes.
+- `redundant_using_join_notes.md`: notes for redundant-column handling around `JOIN ... USING` / `NATURAL JOIN`.

docs/note/planner/redundant_using_join_notes.md

@@ -0,0 +1,38 @@
+# Redundant USING/NATURAL JOIN Notes
+
+## 2026-03-09 - Qualified redundant-column remap must preserve projection identity and predicate type semantics
+
+Background:
+- Issue #66272 originally required remapping redundant `JOIN ... USING` / `NATURAL JOIN` columns so later planner phases do not keep an unresolvable redundant-side column.
+- The root cause is that planner name resolution and executable join output use different column views:
+  - `FullSchema`/`FullNames` still contain the redundant side for qualified-name lookup,
+  - `Join.Schema()`/`OutputNames()` only keep the canonical visible output column.
+- A qualified predicate such as `t3.id = 10` could therefore bind to the redundant side during name resolution, then survive into later optimization even though that redundant column no longer exists in `Join.Schema()`.
+- Two follow-up review findings showed the first fix was too broad:
+  - projection metadata for `SELECT t_right.col` could be mislabeled as the canonical visible side,
+  - `WHERE/HAVING` remap could silently change predicate semantics when the redundant and visible columns had different types.
+
+Key takeaways:
+- Projection naming and predicate remapping have different correctness constraints.
+- For projection metadata, keep the original redundant-side `FullNames` entry so `ResultField` table/original-table metadata still matches the selected column.
+- For predicate remapping, only reuse the canonical visible column when the join is an inner join and both sides have identical `RetType`.
+- Outer joins must not reuse the same remap because null-preserving side semantics are not interchangeable.
+- DML must not reuse the coalesced-output mapping because `UPDATE`/`DELETE` restore the join schema to merged child outputs after `USING`/`NATURAL JOIN` coalescing.
+
+Implementation choice:
+- `coalesceCommonColumns` records `redundant column -> canonical visible output` mappings only for the `SELECT`-style coalesced join output that survives in normal query paths.
+- `findColFromNaturalUsingJoin` now reads the redundant-side identity from `FullSchema`/`FullNames` instead of `ResolveRedundantColumn`.
+- `LogicalJoin.ResolveRedundantColumn` now returns a mapped column only when the redundant-side and visible-side `RetType` values are equal.
+- `expression_rewriter` and `havingWindowAndOrderbyExprResolver` only remap qualified redundant base-table columns for inner joins.
+- `UPDATE`/`DELETE` explicitly skip redundant-column remap because the final DML schema is reset to merged child outputs; a mapping captured from the temporary coalesced output would become stale and could point to the wrong side.
+
+Regression coverage:
+- `SELECT t3.id FROM t1 JOIN t3 USING(id)` verifies result-field metadata still reports `t3`.
+- Mixed-type `VARCHAR`/`INT` `USING(id)` with `WHERE t_mixed_r.id = '01a'` verifies qualified predicates keep right-side integer semantics.
+- `UPDATE ... JOIN ... USING(id)` and `DELETE ... JOIN ... USING(id)` verify DML binding still follows merged-schema semantics.
+- `LEFT JOIN`/`RIGHT JOIN` null-side checks verify outer joins do not incorrectly reuse inner-join remapping.
+
+Validation commands:
+- `make bazel_prepare`
+- `go test ./pkg/planner/core/casetest/schema -run ^TestSchemaCannotFindColumnRegression$ -tags=intest,deadlock`
+- `make lint`

pkg/planner/core/casetest/join/join_test.go

@@ -204,7 +204,6 @@ func TestKeepingJoinKeys(t *testing.T) {
 func TestJoinRegression(t *testing.T) {
 	testkit.RunTestUnderCascades(t, func(t *testing.T, tk *testkit.TestKit, cascades, caller string) {
 		tk.MustExec("use test")
-
 		tk.MustExec(`CREATE TABLE t0(c0 BLOB);`)
 		tk.MustExec(`CREATE definer='root'@'localhost' VIEW v0(c0) AS SELECT NULL FROM t0 GROUP BY NULL;`)
 		tk.MustExec(`SELECT t0.c0 FROM t0 NATURAL JOIN v0 WHERE v0.c0 LIKE v0.c0;`) // no error

pkg/planner/core/casetest/schema/BUILD.bazel

@@ -0,0 +1,19 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_test")
+
+go_test(
+    name = "schema_test",
+    timeout = "short",
+    srcs = [
+        "cannot_find_column_test.go",
+        "main_test.go",
+    ],
+    data = glob(["testdata/**"]),
+    flaky = True,
+    deps = [
+        "//pkg/testkit",
+        "//pkg/testkit/testdata",
+        "//pkg/testkit/testmain",
+        "//pkg/testkit/testsetup",
+        "@org_uber_go_goleak//:goleak",
+    ],
+)

pkg/planner/core/casetest/schema/cannot_find_column_test.go

@@ -0,0 +1,159 @@
+// Copyright 2026 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package schema
+
+import (
+	"testing"
+
+	"github.com/pingcap/tidb/pkg/testkit"
+	"github.com/pingcap/tidb/pkg/testkit/testdata"
+)
+
+func TestSchemaCannotFindColumnRegression(t *testing.T) {
+	testkit.RunTestUnderCascades(t, func(t *testing.T, tk *testkit.TestKit, cascades, caller string) {
+		tk.MustExec("use test")
+		tk.MustExec(`drop table if exists t1, t3, t4`)
+		tk.MustExec(`create table t1 (
+  id bigint primary key,
+  left_v bigint not null
+)`)
+		tk.MustExec(`create table t3 (
+  id bigint primary key,
+  right_v bigint not null
+)`)
+		tk.MustExec(`create table t4 (
+  id bigint primary key,
+  right_v bigint not null,
+  flag tinyint not null
+)`)
+		tk.MustExec("insert into t1 values (10, 93)")
+		tk.MustExec("insert into t3 values (10, 749), (20, 749), (30, 1000)")
+		tk.MustExec("insert into t4 values (10, 749, 1), (20, 749, 0), (30, 1000, 1)")
+
+		var input []string
+		var output []struct {
+			SQL    string
+			Plan   []string
+			Result []string
+		}
+		suite := GetSchemaSuiteData()
+		suite.LoadTestCases(t, &input, &output, cascades, caller)
+		for i, sql := range input {
+			testdata.OnRecord(func() {
+				planRows := testdata.ConvertRowsToStrings(tk.MustQuery("explain format='brief' " + sql).Rows())
+				if len(planRows) == 0 {
+					t.Fatalf("empty plan for sql: %s", sql)
+				}
+				output[i].SQL = sql
+				output[i].Plan = planRows
+				output[i].Result = testdata.ConvertRowsToStrings(tk.MustQuery(sql).Rows())
+			})
+			tk.MustQuery("explain format='brief' " + sql).Check(testkit.Rows(output[i].Plan...))
+			tk.MustQuery(sql).Check(testkit.Rows(output[i].Result...))
+		}
+		tk.MustQuery("SELECT /* issue:66272-nested */ t1.id FROM t1 JOIN t3 USING(id) JOIN t4 ON t4.id = t1.id WHERE t3.id >= 10 AND t3.id <= 20 AND t1.left_v = 93 AND t4.flag = 1").Check(testkit.Rows(
+			"10",
+		))
+		tk.MustQuery("SELECT /* issue:66272-having */ id FROM t1 JOIN t3 USING(id) GROUP BY id HAVING t3.id = 10").Check(testkit.Rows(
+			"10",
+		))
+		tk.MustQuery("SELECT /* issue:66272-orderby */ t1.id FROM t1 JOIN t3 USING(id) WHERE t3.id = 10 ORDER BY t3.id").Check(testkit.Rows(
+			"10",
+		))
+		tk.MustQuery("SELECT /* issue:66272-all */ id AS t0_id FROM t1 JOIN t3 USING(id) WHERE (((t3.right_v = 749) AND (t3.id = 10)) AND (t1.left_v = 93)) AND (t3.right_v = ALL (SELECT t3.right_v FROM t3 WHERE t3.right_v = 749))").Check(testkit.Rows(
+			"10",
+		))
+		stmtID, _, fields, err := tk.Session().PrepareStmt("SELECT /* issue:66272-metadata */ t3.id FROM t1 JOIN t3 USING(id)")
+		tk.RequireNoError(err)
+		defer func() {
+			tk.RequireNoError(tk.Session().DropPreparedStmt(stmtID))
+		}()
+		tk.RequireEqual(1, len(fields))
+		tk.RequireEqual("t3", fields[0].Table.Name.O)
+		tk.RequireEqual("t3", fields[0].TableAsName.O)
+		tk.RequireEqual("id", fields[0].Column.Name.O)
+		tk.RequireEqual("id", fields[0].ColumnAsName.O)
+		tk.MustQuery("SELECT /* issue:66272-metadata */ t3.id FROM t1 JOIN t3 USING(id)").Check(testkit.Rows(
+			"10",
+		))
+
+		tk.MustExec("drop table if exists t_mixed_l, t_mixed_r")
+		tk.MustExec("create table t_mixed_l (id varchar(10) primary key, left_v int not null)")
+		tk.MustExec("create table t_mixed_r (id int primary key, right_v int not null)")
+		tk.MustExec("insert into t_mixed_l values ('01', 10), ('02', 20)")
+		tk.MustExec("insert into t_mixed_r values (1, 100), (2, 200)")
+		tk.MustQuery("SELECT /* issue:66272-type-safe */ t_mixed_r.id FROM t_mixed_l JOIN t_mixed_r USING(id) WHERE t_mixed_r.id = '01a'").Check(testkit.Rows(
+			"1",
+		))
+
+		tk.MustExec("drop table if exists t_up_l, t_up_r")
+		tk.MustExec("create table t_up_l (id int primary key, a int not null)")
+		tk.MustExec("create table t_up_r (id int primary key)")
+		tk.MustExec("insert into t_up_l values (1, 2), (2, 100), (3, 300)")
+		tk.MustExec("insert into t_up_r values (2), (3)")
+		tk.MustExec("update t_up_l join t_up_r using(id) set t_up_l.a = t_up_l.a + 1000 where t_up_r.id = 2")
+		tk.MustQuery("select id, a from t_up_l order by id").Check(testkit.Rows(
+			"1 2",
+			"2 1100",
+			"3 300",
+		))
+
+		tk.MustExec("drop table if exists t_del_l, t_del_r")
+		tk.MustExec("create table t_del_l (id int primary key, a int not null)")
+		tk.MustExec("create table t_del_r (id int primary key)")
+		tk.MustExec("insert into t_del_l values (1, 2), (2, 9), (3, 2)")
+		tk.MustExec("insert into t_del_r values (2), (3)")
+		tk.MustExec("delete t_del_l from t_del_l join t_del_r using(id) where t_del_r.id = 2")
+		tk.MustQuery("select id, a from t_del_l order by id").Check(testkit.Rows(
+			"1 2",
+			"3 2",
+		))
+
+		tk.MustExec("drop table if exists t_ru_l, t_ru_r")
+		tk.MustExec("create table t_ru_l (id int primary key, a int not null)")
+		tk.MustExec("create table t_ru_r (id int primary key)")
+		tk.MustExec("insert into t_ru_l values (1, 2), (2, 100), (3, 300)")
+		tk.MustExec("insert into t_ru_r values (2), (4)")
+		tk.MustExec("update t_ru_l right join t_ru_r using(id) set t_ru_l.a = t_ru_l.a + 1000 where t_ru_r.id = 2")
+		tk.MustQuery("select id, a from t_ru_l order by id").Check(testkit.Rows(
+			"1 2",
+			"2 1100",
+			"3 300",
+		))
+
+		tk.MustExec("drop table if exists t_rd_l, t_rd_r")
+		tk.MustExec("create table t_rd_l (id int primary key, a int not null)")
+		tk.MustExec("create table t_rd_r (id int primary key)")
+		tk.MustExec("insert into t_rd_l values (1, 2), (2, 9), (3, 2)")
+		tk.MustExec("insert into t_rd_r values (2), (4)")
+		tk.MustExec("delete t_rd_l from t_rd_l right join t_rd_r using(id) where t_rd_r.id = 2")
+		tk.MustQuery("select id, a from t_rd_l order by id").Check(testkit.Rows(
+			"1 2",
+			"3 2",
+		))
+
+		tk.MustExec("drop table if exists t_outer_l, t_outer_r")
+		tk.MustExec("create table t_outer_l (id int primary key, a int not null)")
+		tk.MustExec("create table t_outer_r (id int primary key)")
+		tk.MustExec("insert into t_outer_l values (1, 10), (2, 20)")
+		tk.MustExec("insert into t_outer_r values (2), (3)")
+		tk.MustQuery("select count(*) from t_outer_l left join t_outer_r using(id) where t_outer_r.id is null").Check(testkit.Rows(
+			"1",
+		))
+		tk.MustQuery("select count(*) from t_outer_l right join t_outer_r using(id) where t_outer_l.id is null").Check(testkit.Rows(
+			"1",
+		))
+	})
+}

pkg/planner/core/casetest/schema/main_test.go

@@ -0,0 +1,53 @@
+// Copyright 2026 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package schema
+
+import (
+	"flag"
+	"testing"
+
+	"github.com/pingcap/tidb/pkg/testkit/testdata"
+	"github.com/pingcap/tidb/pkg/testkit/testmain"
+	"github.com/pingcap/tidb/pkg/testkit/testsetup"
+	"go.uber.org/goleak"
+)
+
+var testDataMap = make(testdata.BookKeeper)
+
+func TestMain(m *testing.M) {
+	testsetup.SetupForCommonTest()
+	flag.Parse()
+	testDataMap.LoadTestSuiteData("testdata", "cannot_find_column_suite", true)
+	opts := []goleak.Option{
+		goleak.IgnoreTopFunction("github.com/golang/glog.(*fileSink).flushDaemon"),
+		goleak.IgnoreTopFunction("github.com/bazelbuild/rules_go/go/tools/bzltestutil.RegisterTimeoutHandler.func1"),
+		goleak.IgnoreTopFunction("github.com/lestrrat-go/httprc.runFetchWorker"),
+		goleak.IgnoreTopFunction("go.etcd.io/etcd/client/pkg/v3/logutil.(*MergeLogger).outputLoop"),
+		goleak.IgnoreTopFunction("gopkg.in/natefinch/lumberjack%2ev2.(*Logger).millRun"),
+		goleak.IgnoreTopFunction("github.com/tikv/client-go/v2/txnkv/transaction.keepAlive"),
+		goleak.IgnoreTopFunction("go.opencensus.io/stats/view.(*worker).start"),
+	}
+
+	callback := func(i int) int {
+		testDataMap.GenerateOutputIfNeeded()
+		return i
+	}
+
+	goleak.VerifyTestMain(testmain.WrapTestingM(m, callback), opts...)
+}
+
+func GetSchemaSuiteData() testdata.TestData {
+	return testDataMap["cannot_find_column_suite"]
+}

pkg/planner/core/casetest/schema/testdata/cannot_find_column_suite_in.json

@@ -0,0 +1,8 @@
+[
+  {
+    "name": "TestSchemaCannotFindColumnRegression",
+    "cases": [
+      "SELECT /* issue:66272 */ id AS t0_id FROM t1 JOIN t3 USING (id) WHERE (((t3.right_v = 749) AND (t3.id = 10)) AND (t1.left_v = 93)) AND (t3.right_v = ALL (SELECT t3.right_v AS c0 FROM t3 WHERE t3.right_v = 749))"
+    ]
+  }
+]

pkg/planner/core/casetest/schema/testdata/cannot_find_column_suite_out.json

@@ -0,0 +1,26 @@
+[
+  {
+    "Name": "TestSchemaCannotFindColumnRegression",
+    "Cases": [
+      {
+        "SQL": "SELECT /* issue:66272 */ id AS t0_id FROM t1 JOIN t3 USING (id) WHERE (((t3.right_v = 749) AND (t3.id = 10)) AND (t1.left_v = 93)) AND (t3.right_v = ALL (SELECT t3.right_v AS c0 FROM t3 WHERE t3.right_v = 749))",
+        "Plan": [
+          "HashJoin 0.80 root  CARTESIAN inner join",
+          "├─MergeJoin(Build) 1.00 root  inner join, left key:test.t1.id, right key:test.t3.id",
+          "│ ├─Selection(Build) 1.00 root  eq(test.t3.right_v, 749)",
+          "│ │ └─Point_Get 1.00 root table:t3 handle:10",
+          "│ └─Selection(Probe) 1.00 root  eq(test.t1.left_v, 93)",
+          "│   └─Point_Get 1.00 root table:t1 handle:10",
+          "└─Selection(Probe) 0.80 root  or(and(le(Column#11, 1), and(eq(Column#10, 749), if(ne(Column#12, 0), NULL, 1))), eq(Column#13, 0))",
+          "  └─StreamAgg 1.00 root  funcs:max(test.t3.right_v)->Column#10, funcs:count(distinct test.t3.right_v)->Column#11, funcs:sum(0)->Column#12, funcs:count(1)->Column#13",
+          "    └─TableReader 10.00 root  data:Selection",
+          "      └─Selection 10.00 cop[tikv]  eq(test.t3.right_v, 749)",
+          "        └─TableFullScan 10000.00 cop[tikv] table:t3 keep order:false, stats:pseudo"
+        ],
+        "Result": [
+          "10"
+        ]
+      }
+    ]
+  }
+]

pkg/planner/core/casetest/schema/testdata/cannot_find_column_suite_xut.json

@@ -0,0 +1,26 @@
+[
+  {
+    "Name": "TestSchemaCannotFindColumnRegression",
+    "Cases": [
+      {
+        "SQL": "SELECT /* issue:66272 */ id AS t0_id FROM t1 JOIN t3 USING (id) WHERE (((t3.right_v = 749) AND (t3.id = 10)) AND (t1.left_v = 93)) AND (t3.right_v = ALL (SELECT t3.right_v AS c0 FROM t3 WHERE t3.right_v = 749))",
+        "Plan": [
+          "HashJoin 0.80 root  CARTESIAN inner join",
+          "├─MergeJoin(Build) 1.00 root  inner join, left key:test.t1.id, right key:test.t3.id",
+          "│ ├─Selection(Build) 1.00 root  eq(test.t3.right_v, 749)",
+          "│ │ └─Point_Get 1.00 root table:t3 handle:10",
+          "│ └─Selection(Probe) 1.00 root  eq(test.t1.left_v, 93)",
+          "│   └─Point_Get 1.00 root table:t1 handle:10",
+          "└─Selection(Probe) 0.80 root  or(and(le(Column#11, 1), and(eq(Column#10, 749), if(ne(Column#12, 0), NULL, 1))), eq(Column#13, 0))",
+          "  └─StreamAgg 1.00 root  funcs:max(test.t3.right_v)->Column#10, funcs:count(distinct test.t3.right_v)->Column#11, funcs:sum(0)->Column#12, funcs:count(1)->Column#13",
+          "    └─TableReader 10.00 root  data:Selection",
+          "      └─Selection 10.00 cop[tikv]  eq(test.t3.right_v, 749)",
+          "        └─TableFullScan 10000.00 cop[tikv] table:t3 keep order:false, stats:pseudo"
+        ],
+        "Result": [
+          "10"
+        ]
+      }
+    ]
+  }
+]