ddlPuller (ticdc): fix rename table ddl (#9521) (#9537)

ref #9488

cdc/entry/schema/snapshot.go

@@ -85,6 +85,11 @@ func (s *Snapshot) FillSchemaName(job *timodel.Job) error {
 		// DDLs on multiple schema or tables, ignore them.
 		return nil
 	}
+	if job.Type == timodel.ActionRenameTable && job.SchemaName != "" {
+		// DDL on single table with schema name, ignore it.
+		return nil
+	}
+
 	if job.Type == timodel.ActionCreateSchema ||
 		job.Type == timodel.ActionDropSchema {
 		job.SchemaName = job.BinlogInfo.DBInfo.Name.O
@@ -429,6 +434,12 @@ func (s *Snapshot) DoHandleDDL(job *timodel.Job) error {
 		if err != nil {
 			return errors.Trace(err)
 		}
+		// If it a rename table job and the schema does not exist,
+		// there is no need to create the table, since this table
+		// will not be replicated in the future.
+		if _, ok := s.inner.schemaByID(job.SchemaID); !ok {
+			return nil
+		}
 		// create table
 		err = s.inner.createTable(getWrapTableInfo(job), job.BinlogInfo.FinishedTS)
 		if err != nil {

cdc/puller/ddl_puller.go

@@ -124,7 +124,11 @@ func (p *ddlJobPullerImpl) Run(ctx context.Context) error {
 					if err != nil {
 						return errors.Trace(err)
 					}
-					log.Info("handle job", zap.Stringer("job", job), zap.Bool("skip", skip))
+					log.Info("handle ddl job",
+						zap.String("namespace", p.changefeedID.Namespace),
+						zap.String("changefeed", p.changefeedID.ID),
+						zap.String("query", job.Query),
+						zap.Stringer("job", job), zap.Bool("skip", skip))
 					if skip {
 						continue
 					}
@@ -332,9 +336,23 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 		return false, nil
 	}
 
+	defer func() {
+		if skip && err == nil {
+			log.Info("ddl job schema or table does not match, discard it",
+				zap.String("namespace", p.changefeedID.Namespace),
+				zap.String("changefeed", p.changefeedID.ID),
+				zap.String("schema", job.SchemaName),
+				zap.String("table", job.TableName),
+				zap.String("query", job.Query),
+				zap.String("job", job.String()))
+			p.metricDiscardedDDLCounter.Inc()
+		}
+	}()
+
 	snap := p.schemaStorage.GetLastSnapshot()
 	// Do this first to fill the schema name to its origin schema name.
 	if err := snap.FillSchemaName(job); err != nil {
+		log.Info("failed to fill schema name for ddl job", zap.Error(err))
 		// If we can't find a job's schema, check if it's been filtered.
 		if p.filter.ShouldIgnoreTable(job.SchemaName, job.TableName) ||
 			p.filter.ShouldDiscardDDL(job.Type, job.SchemaName, job.TableName) {
@@ -366,6 +384,13 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 			return true, errors.Trace(err)
 		}
 	case timodel.ActionRenameTable:
+		log.Info("rename table ddl job",
+			zap.Int64("newSchemaID", job.SchemaID),
+			zap.String("newSchemaName", job.SchemaName),
+			zap.Int64("tableID", job.TableID),
+			zap.String("oldTableName", job.BinlogInfo.TableInfo.Name.O),
+			zap.String("newTableName", job.TableName),
+		)
 		oldTable, ok := snap.PhysicalTableByID(job.TableID)
 		if !ok {
 			// 1. If we can not find the old table, and the new table name is in filter rule, return error.
@@ -374,10 +399,14 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 			}
 			skip = true
 		} else {
+			log.Info("rename table ddl job",
+				zap.String("oldTableName", oldTable.TableName.Table),
+				zap.String("oldSchemaName", oldTable.TableName.Schema))
 			// 2. If we can find the preTableInfo, we filter it by the old table name.
-			skip = p.filter.ShouldDiscardDDL(job.Type, job.SchemaName, oldTable.Name.O)
+			skipByOldTableName := p.filter.ShouldDiscardDDL(job.Type, oldTable.TableName.Schema, oldTable.TableName.Table)
+			skipByNewTableName := p.filter.ShouldDiscardDDL(job.Type, job.SchemaName, job.BinlogInfo.TableInfo.Name.O)
 			// 3. If its old table name is not in filter rule, and its new table name in filter rule, return error.
-			if skip && !p.filter.ShouldDiscardDDL(job.Type, job.SchemaName, job.BinlogInfo.TableInfo.Name.O) {
+			if skipByOldTableName && !skipByNewTableName {
 				return true, cerror.ErrSyncRenameTableFailed.GenWithStackByArgs(job.TableID, job.Query)
 			}
 		}
@@ -390,14 +419,6 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 	}
 
 	if skip {
-		log.Info("ddl job schema or table does not match, discard it",
-			zap.String("namespace", p.changefeedID.Namespace),
-			zap.String("changefeed", p.changefeedID.ID),
-			zap.String("schema", job.SchemaName),
-			zap.String("table", job.TableName),
-			zap.String("query", job.Query),
-			zap.String("job", job.String()))
-		p.metricDiscardedDDLCounter.Inc()
 		return true, nil
 	}
 

cdc/puller/ddl_puller_test.go

@@ -166,6 +166,9 @@ func TestHandleRenameTable(t *testing.T) {
 		"test1.t66",
 		"test1.t99",
 		"test1.t100",
+		"test1.t20230808",
+		"test1.t202308081",
+		"test1.t202308082",
 
 		"test2.t4",
 
@@ -321,6 +324,20 @@ func TestHandleRenameTable(t *testing.T) {
 		mockPuller.appendResolvedTs(job.BinlogInfo.FinishedTS + 1)
 		waitResolvedTs(t, ddlJobPuller, job.BinlogInfo.FinishedTS+1)
 
+		job = helper.DDL2Job("create table test1.t20230808 (id int)")
+		mockPuller.appendDDL(job)
+		mockPuller.appendResolvedTs(job.BinlogInfo.FinishedTS + 1)
+		waitResolvedTs(t, ddlJobPuller, job.BinlogInfo.FinishedTS+1)
+
+		job = helper.DDL2Job("create table test1.t202308081 (id int)")
+		mockPuller.appendDDL(job)
+		mockPuller.appendResolvedTs(job.BinlogInfo.FinishedTS + 1)
+		waitResolvedTs(t, ddlJobPuller, job.BinlogInfo.FinishedTS+1)
+
+		job = helper.DDL2Job("create table test1.t202308082 (id int)")
+		mockPuller.appendDDL(job)
+		mockPuller.appendResolvedTs(job.BinlogInfo.FinishedTS + 1)
+		waitResolvedTs(t, ddlJobPuller, job.BinlogInfo.FinishedTS+1)
 		// since test1.99 in filter rule, we replicate it
 		job = helper.DDL2Job("rename table test1.t99 to test1.t999")
 		skip, err := ddlJobPullerImpl.handleJob(job)
@@ -341,6 +358,22 @@ func TestHandleRenameTable(t *testing.T) {
 		skip, err = ddlJobPullerImpl.handleJob(job)
 		require.NoError(t, err)
 		require.True(t, skip)
+
+		// since test1.t20230808 is  in filter rule, replicate it
+		// ref: https://github.com/pingcap/tiflow/issues/9488
+		job = helper.DDL2Job("rename table test1.t20230808 to ignore1.ignore")
+		skip, err = ddlJobPullerImpl.handleJob(job)
+		require.NoError(t, err)
+		require.False(t, skip)
+
+		// FIXME(dongmen): since test1.t202308081 and test1.t202308082 are in filter rule, it should be replicated
+		// but now it will throw an error since schema ignore1 are not in schemaStorage
+		// ref: https://github.com/pingcap/tiflow/issues/9488
+		job = helper.DDL2Job("rename table test1.t202308081 to ignore1.ignore1, test1.t202308082 to ignore1.dongmen")
+		skip, err = ddlJobPullerImpl.handleJob(job)
+		require.NotNil(t, err)
+		require.True(t, skip)
+		require.Contains(t, err.Error(), "ErrSnapshotSchemaNotFound")
 	}
 }