ddl_puller(ticdc): discard DDLs that irrelevant with changefeed

cdc/entry/mounter_test.go

@@ -270,7 +270,7 @@ func testMounterDisableOldValue(t *testing.T, tc struct {
 
 	jobs, err := getAllHistoryDDLJob(store)
 	require.Nil(t, err)
-	scheamStorage, err := NewSchemaStorage(nil, 0, nil, false, dummyChangeFeedID)
+	scheamStorage, err := NewSchemaStorage(nil, 0, false, dummyChangeFeedID)
 	require.Nil(t, err)
 	for _, job := range jobs {
 		err := scheamStorage.HandleDDLJob(job)
@@ -993,7 +993,7 @@ func TestDecodeEventIgnoreRow(t *testing.T) {
 	ver, err := helper.Storage().CurrentVersion(oracle.GlobalTxnScope)
 	require.Nil(t, err)
 	schemaStorage, err := NewSchemaStorage(helper.GetCurrentMeta(),
-		ver.Ver, filter, false, cfID)
+		ver.Ver, false, cfID)
 	require.Nil(t, err)
 	// apply ddl to schemaStorage
 	for _, ddl := range ddls {

cdc/entry/schema/snapshot.go

@@ -914,7 +914,7 @@ func (s *snapshot) renameTables(job *timodel.Job, currentTs uint64) error {
 		if !ok {
 			return cerror.ErrSnapshotSchemaNotFound.GenWithStackByArgs(newSchemaIDs[i])
 		}
-		newSchemaName := newSchema.Name.L
+		newSchemaName := newSchema.Name.O
 		tbInfo := model.WrapTableInfo(newSchemaIDs[i], newSchemaName, job.BinlogInfo.FinishedTS, tableInfo)
 		err = s.createTable(tbInfo, currentTs)
 		if err != nil {
@@ -937,7 +937,6 @@ func (s *snapshot) iterTables(includeIneligible bool, f func(i *model.TableInfo)
 		}
 		return true
 	})
-	return
 }
 
 func (s *snapshot) iterPartitions(includeIneligible bool, f func(id int64, i *model.TableInfo)) {

cdc/entry/schema_storage.go

@@ -27,7 +27,6 @@ import (
 	"github.com/pingcap/tiflow/cdc/entry/schema"
 	"github.com/pingcap/tiflow/cdc/model"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
-	"github.com/pingcap/tiflow/pkg/filter"
 	"github.com/pingcap/tiflow/pkg/retry"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
@@ -58,15 +57,14 @@ type schemaStorageImpl struct {
 	resolvedTs    uint64
 	schemaVersion int64
 
-	filter         filter.Filter
 	forceReplicate bool
 
 	id model.ChangeFeedID
 }
 
 // NewSchemaStorage creates a new schema storage
 func NewSchemaStorage(
-	meta *timeta.Meta, startTs uint64, filter filter.Filter,
+	meta *timeta.Meta, startTs uint64,
 	forceReplicate bool, id model.ChangeFeedID,
 ) (SchemaStorage, error) {
 	var (
@@ -93,7 +91,6 @@ func NewSchemaStorage(
 	schema := &schemaStorageImpl{
 		snaps:          []*schema.Snapshot{snap},
 		resolvedTs:     startTs,
-		filter:         filter,
 		forceReplicate: forceReplicate,
 		id:             id,
 		schemaVersion:  version,
@@ -269,12 +266,5 @@ func (s *schemaStorageImpl) skipJob(job *timodel.Job) bool {
 		zap.String("DDL", job.Query), zap.Stringer("job", job),
 		zap.String("namespace", s.id.Namespace),
 		zap.String("changefeed", s.id.ID))
-	if s.filter != nil && s.filter.ShouldDiscardDDL(job.Type) {
-		log.Info("discard DDL",
-			zap.Int64("jobID", job.ID), zap.String("DDL", job.Query),
-			zap.String("namespace", s.id.Namespace),
-			zap.String("changefeed", s.id.ID))
-		return true
-	}
 	return !job.IsSynced() && !job.IsDone()
 }

cdc/entry/schema_storage_test.go

@@ -540,7 +540,7 @@ func TestMultiVersionStorage(t *testing.T) {
 	}
 
 	jobs = append(jobs, job)
-	storage, err := NewSchemaStorage(nil, 0, nil, false, dummyChangeFeedID)
+	storage, err := NewSchemaStorage(nil, 0, false, dummyChangeFeedID)
 	require.Nil(t, err)
 	for _, job := range jobs {
 		err := storage.HandleDDLJob(job)
@@ -856,7 +856,7 @@ func TestSchemaStorage(t *testing.T) {
 
 		jobs, err := getAllHistoryDDLJob(store)
 		require.Nil(t, err)
-		schemaStorage, err := NewSchemaStorage(nil, 0, nil, false, dummyChangeFeedID)
+		schemaStorage, err := NewSchemaStorage(nil, 0, false, dummyChangeFeedID)
 		require.Nil(t, err)
 		for _, job := range jobs {
 			err := schemaStorage.HandleDDLJob(job)

cdc/entry/schema_test_helper.go

@@ -14,6 +14,8 @@
 package entry
 
 import (
+	"encoding/json"
+	"strings"
 	"testing"
 
 	ticonfig "github.com/pingcap/tidb/config"
@@ -64,7 +66,43 @@ func (s *SchemaTestHelper) DDL2Job(ddl string) *timodel.Job {
 	jobs, err := tiddl.GetLastNHistoryDDLJobs(s.GetCurrentMeta(), 1)
 	require.Nil(s.t, err)
 	require.Len(s.t, jobs, 1)
-	return jobs[0]
+	res := jobs[0]
+	if res.Type != timodel.ActionRenameTables {
+		return res
+	}
+
+	// the RawArgs field in job fetched from tidb snapshot meta is incorrent,
+	// so we manually construct `job.RawArgs` to do the workaround.
+	// we assume the old schema name is same as the new schema name here.
+	// for example, "ALTER TABLE RENAME test.t1 TO test.t1, test.t2 to test.t22", schema name is "test"
+	schema := strings.Split(strings.Split(strings.Split(res.Query, ",")[1], " ")[1], ".")[0]
+	tableNum := len(res.BinlogInfo.MultipleTableInfos)
+	oldSchemaIDs := make([]int64, tableNum)
+	for i := 0; i < tableNum; i++ {
+		oldSchemaIDs[i] = res.SchemaID
+	}
+	oldTableIDs := make([]int64, tableNum)
+	for i := 0; i < tableNum; i++ {
+		oldTableIDs[i] = res.BinlogInfo.MultipleTableInfos[i].ID
+	}
+	newTableNames := make([]timodel.CIStr, tableNum)
+	for i := 0; i < tableNum; i++ {
+		newTableNames[i] = res.BinlogInfo.MultipleTableInfos[i].Name
+	}
+	oldSchemaNames := make([]timodel.CIStr, tableNum)
+	for i := 0; i < tableNum; i++ {
+		oldSchemaNames[i] = timodel.NewCIStr(schema)
+	}
+	newSchemaIDs := oldSchemaIDs
+
+	args := []interface{}{
+		oldSchemaIDs, newSchemaIDs,
+		newTableNames, oldTableIDs, oldSchemaNames,
+	}
+	rawArgs, err := json.Marshal(args)
+	require.NoError(s.t, err)
+	res.RawArgs = rawArgs
+	return res
 }
 
 // DDL2Jobs executes the DDL statement and return the corresponding DDL jobs.

cdc/owner/schema.go

@@ -117,19 +117,6 @@ func (s *schemaWrap4Owner) AllTableNames() []model.TableName {
 }
 
 func (s *schemaWrap4Owner) HandleDDL(job *timodel.Job) error {
-	// We use schemaVersion to check if an already-executed DDL job is processed for a second time.
-	// Unexecuted DDL jobs should have largest schemaVersions
-	if job.BinlogInfo.FinishedTS <= s.ddlHandledTs || job.BinlogInfo.SchemaVersion <= s.schemaVersion {
-		log.Warn("job finishTs is less than schema handleTs, discard invalid job",
-			zap.String("namespace", s.id.Namespace),
-			zap.String("changefeed", s.id.ID),
-			zap.Stringer("job", job),
-			zap.Any("ddlHandledTs", s.ddlHandledTs),
-			zap.Int64("schemaVersion", s.schemaVersion),
-			zap.Int64("jobSchemaVersion", job.BinlogInfo.SchemaVersion),
-		)
-		return nil
-	}
 	s.allPhysicalTablesCache = nil
 	err := s.schemaSnapshot.HandleDDL(job)
 	if err != nil {
@@ -187,8 +174,8 @@ func (s *schemaWrap4Owner) parseRenameTables(
 			return nil, cerror.ErrSnapshotSchemaNotFound.GenWithStackByArgs(
 				newSchemaIDs[i])
 		}
-		newSchemaName := newSchema.Name.L
-		oldSchemaName := oldSchemaNames[i].L
+		newSchemaName := newSchema.Name.O
+		oldSchemaName := oldSchemaNames[i].O
 		event := new(model.DDLEvent)
 		preTableInfo, ok := s.schemaSnapshot.PhysicalTableByID(tableInfo.ID)
 		if !ok {

cdc/owner/schema_test.go

@@ -537,5 +537,4 @@ func TestBuildIgnoredDDLJob(t *testing.T) {
 	events, err = schema.BuildDDLEvents(job)
 	require.Nil(t, err)
 	require.Len(t, events, 0)
-	require.Nil(t, schema.HandleDDL(job))
 }

cdc/processor/processor.go

@@ -776,6 +776,7 @@ func (p *processor) createAndDriveSchemaStorage(ctx cdcContext.Context) (entry.S
 		p.upstream.PDClock,
 		checkpointTs,
 		kvCfg,
+		p.changefeed.Info.Config,
 		p.changefeedID,
 	)
 	if err != nil {
@@ -785,7 +786,7 @@ func (p *processor) createAndDriveSchemaStorage(ctx cdcContext.Context) (entry.S
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
-	schemaStorage, err := entry.NewSchemaStorage(meta, checkpointTs, p.filter,
+	schemaStorage, err := entry.NewSchemaStorage(meta, checkpointTs,
 		p.changefeed.Info.Config.ForceReplicate, p.changefeedID)
 	if err != nil {
 		return nil, errors.Trace(err)