plan: fix predicate push down for UnionScan (#7695)

cmd/explaintest/r/explain_easy.result

@@ -352,3 +352,16 @@ create table t(a bigint primary key);
 explain select * from t where a = 1 and a = 2;
 id	count	task	operator info
 TableDual_5	0.00	root	rows:0
+drop table if exists ta, tb;
+create table ta (a varchar(20));
+create table tb (a varchar(20));
+begin;
+insert tb values ('1');
+explain select * from ta where a = 1;
+id	count	task	operator info
+Projection_5	8000.00	root	test.ta.a
+└─Selection_6	8000.00	root	eq(cast(test.ta.a), 1)
+  └─UnionScan_7	10000.00	root	eq(cast(test.ta.a), 1)
+    └─TableReader_9	10000.00	root	data:TableScan_8
+      └─TableScan_8	10000.00	cop	table:ta, range:[-inf,+inf], keep order:false, stats:pseudo
+rollback;

cmd/explaintest/t/explain_easy.test

@@ -70,3 +70,11 @@ explain select * from t where b = 1 and b = 2;
 drop table if exists t;
 create table t(a bigint primary key);
 explain select * from t where a = 1 and a = 2;
+
+drop table if exists ta, tb;
+create table ta (a varchar(20));
+create table tb (a varchar(20));
+begin;
+insert tb values ('1');
+explain select * from ta where a = 1;
+rollback;

executor/union_scan_test.go

@@ -78,3 +78,16 @@ func (s *testSuite) TestDirtyTransaction(c *C) {
 	tk.MustQuery("select * from t use index(idx) where c > 1 and d = 4").Check(testkit.Rows("1 2 3 4"))
 	tk.MustExec("commit")
 }
+
+func (s *testSuite) TestUnionScanWithCastCondition(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("create table ta (a varchar(20))")
+	tk.MustExec("insert ta values ('1'), ('2')")
+	tk.MustExec("create table tb (a varchar(20))")
+	tk.MustExec("begin")
+	tk.MustQuery("select * from ta where a = 1").Check(testkit.Rows("1"))
+	tk.MustExec("insert tb values ('0')")
+	tk.MustQuery("select * from ta where a = 1").Check(testkit.Rows("1"))
+	tk.MustExec("rollback")
+}

plan/rule_predicate_push_down.go

@@ -62,12 +62,13 @@ func (p *LogicalSelection) PredicatePushDown(predicates []expression.Expression)
 
 // PredicatePushDown implements LogicalPlan PredicatePushDown interface.
 func (p *LogicalUnionScan) PredicatePushDown(predicates []expression.Expression) ([]expression.Expression, LogicalPlan) {
-	p.children[0].PredicatePushDown(predicates)
+	retainedPredicates, _ := p.children[0].PredicatePushDown(predicates)
 	p.conditions = make([]expression.Expression, 0, len(predicates))
 	for _, cond := range predicates {
 		p.conditions = append(p.conditions, cond)
 	}
-	return nil, p
+	// The conditions in UnionScan is only used for added rows, so parent Selection should not be removed.
+	return retainedPredicates, p
 }
 
 // PredicatePushDown implements LogicalPlan PredicatePushDown interface.