planner/core,binloginfo: add a projection to fix 'delete from join' schema

planner/core/logical_plan_builder.go

@@ -2328,6 +2328,7 @@ func (b *PlanBuilder) buildDelete(delete *ast.DeleteStmt) (Plan, error) {
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
+	oldSchema := p.Schema()
 
 	if sel.Where != nil {
 		p, err = b.buildSelection(p, sel.Where, nil)
@@ -2350,6 +2351,15 @@ func (b *PlanBuilder) buildDelete(delete *ast.DeleteStmt) (Plan, error) {
 		}
 	}
 
+	// Add a projection for the following case, otherwise the final schema will be the schema of the join.
+	// delete from t where a in (select ...) or b in (select ...)
+	if oldLen := oldSchema.Len(); oldLen != p.Schema().Len() {
+		proj := LogicalProjection{Exprs: expression.Column2Exprs(p.Schema().Columns[:oldLen])}.Init(b.ctx)
+		proj.SetChildren(p)
+		proj.SetSchema(oldSchema.Clone())
+		p = proj
+	}
+
 	var tables []*ast.TableName
 	if delete.Tables != nil {
 		tables = delete.Tables.Tables

planner/core/logical_plan_test.go

@@ -741,6 +741,14 @@ func (s *testPlanSuite) TestPlanBuilder(c *C) {
 			sql:  "select * from t t1 natural join t t2",
 			plan: "Join{DataScan(t1)->DataScan(t2)}->Projection",
 		},
+		{
+			sql: "delete from t where a in (select b from t where c = 666) or b in (select a from t where c = 42)",
+			// Note the Projection before Delete: the final schema should be the schema of
+			// table t rather than Join.
+			// If this schema is not set correctly, table.RemoveRecord would fail when adding
+			// binlog columns, because the schema and data are not consistent.
+			plan: "LeftHashJoin{LeftHashJoin{TableReader(Table(t))->IndexLookUp(Index(t.c_d_e)[[666,666]], Table(t))}(test.t.a,test.t.b)->IndexReader(Index(t.c_d_e)[[42,42]])}(test.t.b,test.t.a)->Sel([or(6_aux_0, 10_aux_0)])->Projection->Delete",
+		},
 	}
 	for _, ca := range tests {
 		comment := Commentf("for %s", ca.sql)

sessionctx/binloginfo/binloginfo_test.go

@@ -121,8 +121,8 @@ func (s *testBinlogSuite) TearDownSuite(c *C) {
 	s.ddl.Stop()
 	s.serv.Stop()
 	os.Remove(s.unixFile)
-	s.store.Close()
 	s.domain.Close()
+	s.store.Close()
 }
 
 func (s *testBinlogSuite) TestBinlog(c *C) {
@@ -419,3 +419,16 @@ func (s *testBinlogSuite) TestPartitionedTable(c *C) {
 		c.Assert(tids[i], Equals, tids[0])
 	}
 }
+
+func (s *testBinlogSuite) TestDeleteSchema(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("CREATE TABLE `b1` (`id` int(11) NOT NULL AUTO_INCREMENT, `job_id` varchar(50) NOT NULL, `split_job_id` varchar(30) DEFAULT NULL, PRIMARY KEY (`id`), KEY `b1` (`job_id`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;")
+	tk.MustExec("CREATE TABLE `b2` (`id` int(11) NOT NULL AUTO_INCREMENT, `job_id` varchar(50) NOT NULL, `batch_class` varchar(20) DEFAULT NULL, PRIMARY KEY (`id`), UNIQUE KEY `bu` (`job_id`)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4")
+	tk.MustExec("insert into b2 (job_id, batch_class) values (2, 'TEST');")
+	tk.MustExec("insert into b1 (job_id) values (2);")
+
+	// This test cover a bug that the final schema and the binlog row inconsistent.
+	// The final schema of this SQL should be the schema of table b1, rather than the schema of join result.
+	tk.MustExec("delete from b1 where job_id in (select job_id from b2 where batch_class = 'TEST') or split_job_id in (select job_id from b2 where batch_class = 'TEST');")
+}