planner: fix join reorder schema mismatch with full schema columns

.gitignore

@@ -48,6 +48,7 @@ MODULE.bazel.lock
 # Testing / coverage / reports
 # Files generated when testing
 out
+/reports/
 test_coverage
 coverage.out
 coverage.txt

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

@@ -6,9 +6,11 @@ go_test(
     srcs = [
         "join_test.go",
         "main_test.go",
+        "reports_test.go",
     ],
+    data = glob(["testdata/**"]),
     flaky = True,
-    shard_count = 5,
+    shard_count = 7,
     deps = [
         "//pkg/config",
         "//pkg/testkit",

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

@@ -145,6 +145,33 @@ func TestJoinWithNullEQ(t *testing.T) {
 	})
 }
 
+func TestCantFindColumnJoinReorder(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 t0, t1, t2, t3, t4;")
+		tk.MustExec("create table t0 (id bigint, k0 varchar(64), k1 varchar(64), k2 int, k3 int, p0 float, p1 tinyint(1), primary key (id));")
+		tk.MustExec("create table t1 (id bigint, k0 varchar(64), d0 bigint, d1 double, primary key (id));")
+		tk.MustExec("create table t2 (id bigint, k1 varchar(64), k0 varchar(64), d0 decimal(12,2), d1 double, primary key (id));")
+		tk.MustExec("create table t3 (id bigint, k2 int, k0 varchar(64), d0 float, d1 date, primary key (id), key idx_id_4 (id));")
+		tk.MustExec("create table t4 (id bigint, k3 int, k0 varchar(64), d0 date, d1 bigint, primary key (id));")
+
+		query := "select distinct length(t0.k3) as c0 from t0 " +
+			"join t2 on ((t0.k0 = t2.k0) and ((t0.k0 = t2.k0) and (t0.k0 != t2.k0))) " +
+			"left join t4 on ((t0.k0 = t4.k0) and not (t0.k0 in ('s0'))) " +
+			"left join t1 on ((t0.k0 = t1.k0) and ((t4.k0 < t1.k0) and (t2.k0 != t1.k0))) " +
+			"left join t3 on ((t0.k0 = t3.k0) and ((t0.k0 <= t3.k0) and (t2.k0 < t4.k0))) " +
+			"where (not (t0.k0 in ('s85','s56','s87')) and not (t2.k0 in ((select t0.k0 as c0 from t0 where (t0.k3 = t0.k3)))));"
+		tk.MustQuery(query).Check(testkit.Rows())
+
+		tk.MustExec("drop table if exists t1, t2, t3;")
+		// Issue: https://github.com/pingcap/tidb/issues/65454
+		tk.MustExec("create table t1 (id bigint not null, hcode text collate utf8mb4_general_ci not null, primary key (id) /*T![clustered_index] CLUSTERED */);")
+		tk.MustExec("create table t2 (id bigint not null, bid bigint not null, cid bigint not null, primary key (id) /*T![clustered_index] CLUSTERED */);")
+		tk.MustExec("create table t3 (id bigint not null, user_id bigint not null, primary key (id) /*T![clustered_index] CLUSTERED */);")
+		tk.MustExec("explain format = 'plan_tree' select (select count(t3.user_id) from t2 left join t1 as cm on t2.cid = cm.id left join t3 on t2.bid = t3.id where cm.hcode = t1.hcode) as tt from t1 where t1.id in (select min(id) from t1);")
+	})
+}
+
 func TestJoinSimplifyCondition(t *testing.T) {
 	testkit.RunTestUnderCascades(t, func(t *testing.T, tk *testkit.TestKit, cascades, caller string) {
 		tk.MustExec("use test;")

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

@@ -0,0 +1,229 @@
+// 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 join
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"regexp"
+	"sort"
+	"strings"
+	"testing"
+
+	"github.com/pingcap/tidb/pkg/testkit"
+	"github.com/stretchr/testify/require"
+)
+
+type reportSummary struct {
+	Error string `json:"error"`
+}
+
+func TestReportsCantFindColumn(t *testing.T) {
+	// issue: 65454
+	testkit.RunTestUnderCascades(t, func(t *testing.T, tk *testkit.TestKit, cascades, caller string) {
+		reportsDir := filepath.Join("testdata", "reports")
+		summaries, err := filepath.Glob(filepath.Join(reportsDir, "case_*", "summary.json"))
+		require.NoError(t, err)
+		require.NotEmpty(t, summaries, "no deterministic report cases found in %s", reportsDir)
+
+		for _, summaryPath := range summaries {
+			summary, err := readReportSummary(summaryPath)
+			require.NoError(t, err)
+			if !strings.Contains(summary.Error, "Can't find column") {
+				continue
+			}
+			caseDir := filepath.Dir(summaryPath)
+			caseName := filepath.Base(caseDir)
+			t.Run(caseName, func(t *testing.T) {
+				schemaSQL := mustReadFile(t, filepath.Join(caseDir, "schema.sql"))
+				insertSQL := mustReadFile(t, filepath.Join(caseDir, "inserts.sql"))
+				caseSQL := mustReadFile(t, filepath.Join(caseDir, "case.sql"))
+
+				dbNames := collectDBNames(schemaSQL, insertSQL, caseSQL)
+				if len(dbNames) == 0 {
+					dbNames = []string{"report_" + strings.ReplaceAll(caseName, "-", "_")}
+				}
+				for _, dbName := range dbNames {
+					tk.MustExec("drop database if exists " + dbName)
+					tk.MustExec("create database " + dbName)
+				}
+				tk.MustExec("use " + dbNames[0])
+
+				if err := runSQLText(t, tk, schemaSQL, false); err != nil {
+					t.Skipf("skip case due to setup error: %v", err)
+				}
+				if err := runSQLText(t, tk, insertSQL, false); err != nil {
+					t.Skipf("skip case due to setup error: %v", err)
+				}
+				_ = runSQLText(t, tk, caseSQL, true)
+
+				for _, dbName := range dbNames {
+					tk.MustExec("drop database if exists " + dbName)
+				}
+			})
+		}
+	})
+}
+
+func readReportSummary(path string) (*reportSummary, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return nil, err
+	}
+	var summary reportSummary
+	if err := json.Unmarshal(data, &summary); err != nil {
+		return nil, err
+	}
+	return &summary, nil
+}
+
+func mustReadFile(t *testing.T, path string) string {
+	data, err := os.ReadFile(path)
+	require.NoError(t, err)
+	return string(data)
+}
+
+func runSQLText(t *testing.T, tk *testkit.TestKit, sqlText string, allowNonCantFind bool) error {
+	for _, stmt := range splitSQL(sqlText) {
+		if err := execSQL(t, tk, stmt, allowNonCantFind); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func execSQL(t *testing.T, tk *testkit.TestKit, stmt string, allowNonCantFind bool) error {
+	sqlText := strings.TrimSpace(stmt)
+	if sqlText == "" {
+		return nil
+	}
+	rs, err := tk.Exec(sqlText)
+	if rs != nil {
+		closeErr := rs.Close()
+		if err == nil && closeErr != nil {
+			return closeErr
+		}
+	}
+	if err == nil {
+		return nil
+	}
+	if strings.Contains(err.Error(), "Can't find column") {
+		require.NoError(t, err, "sql:%s", sqlText)
+		return err
+	}
+	if !allowNonCantFind {
+		return err
+	}
+	return nil
+}
+
+func collectDBNames(sqlTexts ...string) []string {
+	dbs := make(map[string]struct{})
+	nameRe := regexp.MustCompile(`(?i)\bshiro_fuzz[0-9a-zA-Z_]*\b`)
+	for _, sqlText := range sqlTexts {
+		for _, name := range nameRe.FindAllString(sqlText, -1) {
+			dbs[strings.ToLower(name)] = struct{}{}
+		}
+		for _, stmt := range splitSQL(sqlText) {
+			fields := strings.Fields(strings.TrimSpace(stmt))
+			if len(fields) == 0 {
+				continue
+			}
+			switch strings.ToLower(fields[0]) {
+			case "use":
+				if len(fields) >= 2 {
+					name := trimDBName(fields[1])
+					if name != "" {
+						dbs[strings.ToLower(name)] = struct{}{}
+					}
+				}
+			case "create":
+				if len(fields) >= 3 && strings.ToLower(fields[1]) == "database" {
+					nameIdx := 2
+					if len(fields) >= 6 && strings.ToLower(fields[2]) == "if" && strings.ToLower(fields[3]) == "not" && strings.ToLower(fields[4]) == "exists" {
+						nameIdx = 5
+					}
+					if len(fields) > nameIdx {
+						name := trimDBName(fields[nameIdx])
+						if name != "" {
+							dbs[strings.ToLower(name)] = struct{}{}
+						}
+					}
+				}
+			}
+		}
+	}
+	if len(dbs) == 0 {
+		return nil
+	}
+	names := make([]string, 0, len(dbs))
+	for name := range dbs {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+	return names
+}
+
+func trimDBName(name string) string {
+	name = strings.TrimSpace(name)
+	name = strings.TrimSuffix(name, ";")
+	name = strings.Trim(name, "`")
+	return name
+}
+
+func splitSQL(sql string) []string {
+	var res []string
+	var b strings.Builder
+	inSingle, inDouble, inBack := false, false, false
+	escaped := false
+	for _, r := range sql {
+		if escaped {
+			escaped = false
+			b.WriteRune(r)
+			continue
+		}
+		if r == '\\' && (inSingle || inDouble) {
+			escaped = true
+			b.WriteRune(r)
+			continue
+		}
+		switch r {
+		case '\'':
+			if !inDouble && !inBack {
+				inSingle = !inSingle
+			}
+		case '"':
+			if !inSingle && !inBack {
+				inDouble = !inDouble
+			}
+		case '`':
+			if !inSingle && !inDouble {
+				inBack = !inBack
+			}
+		case ';':
+			if !inSingle && !inDouble && !inBack {
+				res = append(res, b.String())
+				b.Reset()
+				continue
+			}
+		}
+		b.WriteRune(r)
+	}
+	if strings.TrimSpace(b.String()) != "" {
+		res = append(res, b.String())
+	}
+	return res
+}

pkg/planner/core/casetest/join/testdata/reports/case_65454/case.sql

@@ -0,0 +1,6 @@
+select distinct length(t0.k3) as c0 from t0
+join t2 on ((t0.k0 = t2.k0) and ((t0.k0 = t2.k0) and (t0.k0 != t2.k0)))
+left join t4 on ((t0.k0 = t4.k0) and not (t0.k0 in ('s0')))
+left join t1 on ((t0.k0 = t1.k0) and ((t4.k0 < t1.k0) and (t2.k0 != t1.k0)))
+left join t3 on ((t0.k0 = t3.k0) and ((t0.k0 <= t3.k0) and (t2.k0 < t4.k0)))
+where (not (t0.k0 in ('s85','s56','s87')) and not (t2.k0 in ((select t0.k0 as c0 from t0 where (t0.k3 = t0.k3)))));

pkg/planner/core/casetest/join/testdata/reports/case_65454/inserts.sql

@@ -0,0 +1,5 @@
+insert into t0 values (1, 's1', 'k1', 1, 3, 1.0, 1);
+insert into t1 values (1, 's1', 1, 1.0);
+insert into t2 values (1, 'k1', 's1', 1.00, 1.0);
+insert into t3 values (1, 1, 's1', 1.0, '2024-01-01');
+insert into t4 values (1, 3, 's1', '2024-01-01', 1);

pkg/planner/core/casetest/join/testdata/reports/case_65454/schema.sql

@@ -0,0 +1,8 @@
+create database if not exists shiro_fuzz_r1;
+use shiro_fuzz_r1;
+drop table if exists t0, t1, t2, t3, t4;
+create table t0 (id bigint, k0 varchar(64), k1 varchar(64), k2 int, k3 int, p0 float, p1 tinyint(1), primary key (id));
+create table t1 (id bigint, k0 varchar(64), d0 bigint, d1 double, primary key (id));
+create table t2 (id bigint, k1 varchar(64), k0 varchar(64), d0 decimal(12,2), d1 double, primary key (id));
+create table t3 (id bigint, k2 int, k0 varchar(64), d0 float, d1 date, primary key (id), key idx_id_4 (id));
+create table t4 (id bigint, k3 int, k0 varchar(64), d0 date, d1 bigint, primary key (id));

pkg/planner/core/casetest/join/testdata/reports/case_65454/summary.json

@@ -0,0 +1 @@
+{"error":"Can't find column test.t4.k0 in schema"}

pkg/planner/core/operator/logicalop/logical_apply.go

@@ -232,7 +232,30 @@ func (la *LogicalApply) CanPullUpAgg() bool {
 	if len(la.EqualConditions)+len(la.LeftConditions)+len(la.RightConditions)+len(la.OtherConditions) > 0 {
 		return false
 	}
-	return len(la.Children()[0].Schema().PKOrUK) > 0
+	children := la.Children()
+	if len(children) == 0 || children[0] == nil {
+		return false
+	}
+	outerSchema := children[0].Schema()
+	if outerSchema == nil {
+		return false
+	}
+	for _, key := range outerSchema.PKOrUK {
+		if len(key) == 0 {
+			continue
+		}
+		allVisible := true
+		for _, keyCol := range key {
+			if keyCol == nil || !outerSchema.Contains(keyCol) {
+				allVisible = false
+				break
+			}
+		}
+		if allVisible {
+			return true
+		}
+	}
+	return false
 }
 
 // DeCorColFromEqExpr checks whether it's an equal condition of form `col = correlated col`. If so we will change the decorrelated