syncer(dm): fix different output format for operate-schema get (#5824)

dm/syncer/schema.go

@@ -16,13 +16,20 @@ package syncer
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"regexp"
 	"strings"
 
 	"github.com/pingcap/tidb-tools/pkg/filter"
 	"github.com/pingcap/tidb/parser/ast"
 	"github.com/pingcap/tidb/parser/format"
 	"github.com/pingcap/tidb/parser/model"
+<<<<<<< HEAD
+=======
+	"github.com/pingcap/tidb/util/filter"
+	"github.com/pingcap/tiflow/dm/pkg/utils"
+	"github.com/pingcap/tiflow/pkg/quotes"
+>>>>>>> 1ba147108 (syncer(dm): fix different output format for operate-schema get (#5824))
 	"go.uber.org/zap"
 
 	"github.com/pingcap/tiflow/dm/dm/config"
@@ -66,9 +73,27 @@ func (s *Syncer) OperateSchema(ctx context.Context, req *pb.OperateWorkerSchemaR
 		}
 		return string(tableListJSON), err
 	case pb.SchemaOp_GetSchema:
+<<<<<<< HEAD
 		// we only try to get schema from schema-tracker now.
 		// in other words, we can not get the schema if any DDL/DML has been replicated, or set a schema previously.
 		return s.schemaTracker.GetCreateTable(ctx, sourceTable)
+=======
+		// when task is paused, schemaTracker is closed. We get the table structure from checkpoint.
+		ti := s.checkpoint.GetTableInfo(req.Database, req.Table)
+		if ti == nil {
+			s.tctx.L().Info("table schema is not in checkpoint, fetch from downstream",
+				zap.String("table", sourceTable.String()))
+			targetTable := s.route(sourceTable)
+			result, err2 := dbconn.GetTableCreateSQL(s.tctx.WithContext(ctx), s.downstreamTrackConn, targetTable.String())
+			result = strings.Replace(result, fmt.Sprintf("CREATE TABLE %s", quotes.QuoteName(targetTable.Name)), fmt.Sprintf("CREATE TABLE %s", quotes.QuoteName(sourceTable.Name)), 1)
+			return utils.CreateTableSQLToOneRow(result), err2
+		}
+
+		result := bytes.NewBuffer(make([]byte, 0, 512))
+		err2 := executor.ConstructResultOfShowCreateTable(s.sessCtx, ti, autoid.Allocators{}, result)
+		return utils.CreateTableSQLToOneRow(result.String()), err2
+
+>>>>>>> 1ba147108 (syncer(dm): fix different output format for operate-schema get (#5824))
 	case pb.SchemaOp_SetSchema:
 		// from source or target need get schema
 		if req.FromSource {

dm/syncer/syncer_test.go

@@ -59,6 +59,12 @@ import (
 	"github.com/pingcap/tidb/parser"
 	"github.com/pingcap/tidb/parser/ast"
 	pmysql "github.com/pingcap/tidb/parser/mysql"
+<<<<<<< HEAD
+=======
+	"github.com/pingcap/tidb/util/filter"
+	regexprrouter "github.com/pingcap/tidb/util/regexpr-router"
+	router "github.com/pingcap/tidb/util/table-router"
+>>>>>>> 1ba147108 (syncer(dm): fix different output format for operate-schema get (#5824))
 	"go.uber.org/zap"
 )
 
@@ -1011,6 +1017,50 @@ func (s *testSyncerSuite) TestRun(c *C) {
 
 	cancel()
 	<-resultCh // wait for the process to finish
+
+	// test OperateSchema starts
+	ctx, cancel = context.WithCancel(context.Background())
+
+	syncer.sessCtx = utils.NewSessionCtx(map[string]string{"time_zone": "UTC"})
+	sourceSchemaFromCheckPoint, err := syncer.OperateSchema(ctx, &pb.OperateWorkerSchemaRequest{Op: pb.SchemaOp_GetSchema, Database: "test_1", Table: "t_1"})
+	c.Assert(err, IsNil)
+
+	syncer.tableRouter = &regexprrouter.RouteTable{}
+	c.Assert(syncer.tableRouter.AddRule(&router.TableRule{
+		SchemaPattern: "test_1",
+		TablePattern:  "t_1",
+		TargetSchema:  "test_1",
+		TargetTable:   "t_2",
+	}), IsNil)
+
+	syncer.checkpoint.(*RemoteCheckPoint).points = make(map[string]map[string]*binlogPoint)
+
+	showTableResultString := "CREATE TABLE `t_2` (\n" +
+		"  `id` int(11) NOT NULL,\n" +
+		"  `name` varchar(24) DEFAULT NULL,\n" +
+		"  PRIMARY KEY (`id`) /*T![clustered_index] NONCLUSTERED */,\n" +
+		"  KEY `index1` (`name`)\n" +
+		") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"
+
+	mock.ExpectQuery("SHOW CREATE TABLE " + "`test_1`.`t_2`").WillReturnRows(
+		sqlmock.NewRows([]string{"Table", "Create Table"}).
+			AddRow("t_2", showTableResultString))
+
+	sourceSchemaFromDownstream, err := syncer.OperateSchema(ctx, &pb.OperateWorkerSchemaRequest{Op: pb.SchemaOp_GetSchema, Database: "test_1", Table: "t_1"})
+	c.Assert(err, IsNil)
+
+	sourceSchemaExpected := "CREATE TABLE `t_1` (" +
+		" `id` int(11) NOT NULL," +
+		" `name` varchar(24) DEFAULT NULL," +
+		" PRIMARY KEY (`id`) /*T![clustered_index] NONCLUSTERED */," +
+		" KEY `index1` (`name`)" +
+		") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"
+	c.Assert(sourceSchemaFromCheckPoint, Equals, sourceSchemaExpected)
+	c.Assert(sourceSchemaFromDownstream, Equals, sourceSchemaExpected)
+
+	cancel()
+	// test OperateSchema ends
+
 	syncer.Close()
 	c.Assert(syncer.isClosed(), IsTrue)