*: support clustered index table produce binlog (pingcap#24135)

heiheipp · May 6, 2021 · 6ef4815 · 6ef4815
1 parent 13ff072
commit 6ef4815
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 27 deletions.
diff --git a/ddl/ddl_api.go b/ddl/ddl_api.go
@@ -1440,11 +1440,6 @@ func buildTableInfo(
 				if isSingleIntPK {
 					tbInfo.PKIsHandle = true
 				} else {
-					hasBinlog := ctx.GetSessionVars().BinlogClient != nil
-					if hasBinlog {
-						msg := mysql.Message("Cannot create clustered index table when the binlog is ON", nil)
-						return nil, dbterror.ClassDDL.NewStdErr(errno.ErrUnsupportedDDLOperation, msg)
-					}
 					tbInfo.IsCommonHandle = true
 					tbInfo.CommonHandleVersion = 1
 				}

diff --git a/sessionctx/binloginfo/binloginfo_test.go b/sessionctx/binloginfo/binloginfo_test.go
@@ -269,25 +269,6 @@ func (s *testBinlogSuite) TestBinlog(c *C) {
 		binlog.MutationType_Insert,
 	})
 
-	// Cannot create common clustered index table when binlog client exists.
-	errMsg := "[ddl:8200]Cannot create clustered index table when the binlog is ON"
-	tk.MustGetErrMsg("create table local_clustered_index (c1 varchar(255) primary key clustered);", errMsg)
-	// Create int clustered index table when binlog client exists.
-	tk.MustExec("create table local_clustered_index (c1 bigint primary key clustered);")
-	tk.MustQuery("select tidb_pk_type from information_schema.tables where table_name = 'local_clustered_index' and table_schema = 'test';").
-		Check(testkit.Rows("CLUSTERED"))
-	tk.MustExec("drop table if exists local_clustered_index;")
-	// Test common clustered index tables will not write binlog.
-	tk.Se.GetSessionVars().BinlogClient = nil
-	tk.MustExec("create table local_clustered_index (c1 varchar(255) primary key clustered);")
-	tk.MustQuery("select tidb_pk_type from information_schema.tables where table_name = 'local_clustered_index' and table_schema = 'test';").
-		Check(testkit.Rows("CLUSTERED"))
-	tk.Se.GetSessionVars().BinlogClient = s.client
-	// This statement should not write binlog.
-	tk.MustExec(`insert into local_clustered_index values ("aaaaaa")`)
-	prewriteVal = getLatestBinlogPrewriteValue(c, pump)
-	c.Assert(len(prewriteVal.Mutations), Equals, 0)
-
 	checkBinlogCount(c, pump)
 
 	pump.mu.Lock()

diff --git a/table/tables/tables.go b/table/tables/tables.go
@@ -409,7 +409,7 @@ func (t *TableCommon) UpdateRecord(ctx context.Context, sctx sessionctx.Context,
 	}
 	memBuffer.Release(sh)
 	if shouldWriteBinlog(sctx, t.meta) {
-		if !t.meta.PKIsHandle {
+		if !t.meta.PKIsHandle && !t.meta.IsCommonHandle {
 			binlogColIDs = append(binlogColIDs, model.ExtraHandleID)
 			binlogOldRow = append(binlogOldRow, types.NewIntDatum(h.IntValue()))
 			binlogNewRow = append(binlogNewRow, types.NewIntDatum(h.IntValue()))
@@ -1029,7 +1029,7 @@ func (t *TableCommon) RemoveRecord(ctx sessionctx.Context, h kv.Handle, r []type
 			colIDs = append(colIDs, col.ID)
 		}
 		var binlogRow []types.Datum
-		if !t.meta.PKIsHandle {
+		if !t.meta.PKIsHandle && !t.meta.IsCommonHandle {
 			colIDs = append(colIDs, model.ExtraHandleID)
 			binlogRow = make([]types.Datum, 0, len(r)+1)
 			binlogRow = append(binlogRow, r...)
@@ -1407,7 +1407,7 @@ func shouldWriteBinlog(ctx sessionctx.Context, tblInfo *model.TableInfo) bool {
 	if ctx.GetSessionVars().BinlogClient == nil {
 		return false
 	}
-	return !ctx.GetSessionVars().InRestrictedSQL && !tblInfo.IsCommonHandle
+	return !ctx.GetSessionVars().InRestrictedSQL
 }
 
 func (t *TableCommon) getMutation(ctx sessionctx.Context) *binlog.TableMutation {