This is an automated cherry-pick of pingcap#43759

Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>

planner/core/issuetest/BUILD.bazel

@@ -6,5 +6,18 @@ go_test(
     srcs = ["planner_issue_test.go"],
     flaky = True,
     race = "on",
+<<<<<<< HEAD
     deps = ["//testkit"],
+=======
+    shard_count = 3,
+    deps = [
+        "//parser",
+        "//planner",
+        "//planner/core",
+        "//testkit",
+        "//testkit/testsetup",
+        "@com_github_stretchr_testify//require",
+        "@org_uber_go_goleak//:goleak",
+    ],
+>>>>>>> 6caadcafc6f (planner: fix correctness of the correlated predicate push down for cte (#43759))
 )

planner/core/issuetest/planner_issue_test.go

@@ -40,3 +40,44 @@ func TestIssue43178(t *testing.T) {
 	// Should not panic
 	tk.MustExec("explain select  /*+ use_index_merge( `aa311c3c` ) */   `aa311c3c`.`43b06e99` as r0 , `aa311c3c`.`6302d8ac` as r1 from `aa311c3c` where IsNull( `aa311c3c`.`b80b3746` ) or not( `aa311c3c`.`57fd8d09` >= '2008' )   order by r0,r1 limit 95")
 }
+<<<<<<< HEAD
+=======
+
+// It's a case for Columns in tableScan and indexScan with double reader
+func TestIssue43461(t *testing.T) {
+	store, domain := testkit.CreateMockStoreAndDomain(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t(a int, b int, c int, index b(b), index b_c(b, c)) partition by hash(a) partitions 4;")
+	tk.MustExec("analyze table t")
+
+	stmt, err := parser.New().ParseOneStmt("select * from t use index(b) where b > 1 order by b limit 1", "", "")
+	require.NoError(t, err)
+
+	p, _, err := planner.Optimize(context.TODO(), tk.Session(), stmt, domain.InfoSchema())
+	require.NoError(t, err)
+	require.NotNil(t, p)
+
+	idxLookUpPlan, ok := p.(*core.PhysicalLimit).Children()[0].(*core.PhysicalProjection).Children()[0].(*core.PhysicalIndexLookUpReader)
+	require.True(t, ok)
+
+	is := idxLookUpPlan.IndexPlans[0].(*core.PhysicalIndexScan)
+	ts := idxLookUpPlan.TablePlans[0].(*core.PhysicalTableScan)
+
+	require.NotEqual(t, is.Columns, ts.Columns)
+}
+
+func TestIssue43645(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+
+	tk.MustExec("use test")
+	tk.MustExec("CREATE TABLE t1(id int,col1 varchar(10),col2 varchar(10),col3 varchar(10));")
+	tk.MustExec("CREATE TABLE t2(id int,col1 varchar(10),col2 varchar(10),col3 varchar(10));")
+	tk.MustExec("INSERT INTO t1 values(1,NULL,NULL,null),(2,NULL,NULL,null),(3,NULL,NULL,null);")
+	tk.MustExec("INSERT INTO t2 values(1,'a','aa','aaa'),(2,'b','bb','bbb'),(3,'c','cc','ccc');")
+
+	rs := tk.MustQuery("WITH tmp AS (SELECT t2.* FROM t2) select (SELECT tmp.col1 FROM tmp WHERE tmp.id=t1.id ) col1, (SELECT tmp.col2 FROM tmp WHERE tmp.id=t1.id ) col2, (SELECT tmp.col3 FROM tmp WHERE tmp.id=t1.id ) col3 from t1;")
+	rs.Sort().Check(testkit.Rows("a aa aaa", "b bb bbb", "c cc ccc"))
+}
+>>>>>>> 6caadcafc6f (planner: fix correctness of the correlated predicate push down for cte (#43759))

planner/core/rule_predicate_push_down.go

@@ -988,13 +988,26 @@ func (p *LogicalCTE) PredicatePushDown(predicates []expression.Expression, _ *lo
 	if !p.isOuterMostCTE {
 		return predicates, p.self
 	}
-	if len(predicates) == 0 {
+	pushedPredicates := make([]expression.Expression, len(predicates))
+	copy(pushedPredicates, predicates)
+	// The filter might change the correlated status of the cte.
+	// We forbid the push down that makes the change for now.
+	// Will support it later.
+	if !p.cte.IsInApply {
+		for i := len(pushedPredicates) - 1; i >= 0; i-- {
+			if len(expression.ExtractCorColumns(pushedPredicates[i])) == 0 {
+				continue
+			}
+			pushedPredicates = append(pushedPredicates[0:i], pushedPredicates[i+1:]...)
+		}
+	}
+	if len(pushedPredicates) == 0 {
 		p.cte.pushDownPredicates = append(p.cte.pushDownPredicates, expression.NewOne())
 		return predicates, p.self
 	}
 	newPred := make([]expression.Expression, 0, len(predicates))
-	for i := range predicates {
-		newPred = append(newPred, predicates[i].Clone())
+	for i := range pushedPredicates {
+		newPred = append(newPred, pushedPredicates[i].Clone())
 		ResolveExprAndReplace(newPred[i], p.cte.ColumnMap)
 	}
 	p.cte.pushDownPredicates = append(p.cte.pushDownPredicates, expression.ComposeCNFCondition(p.ctx, newPred...))